PK ! X . electricitycostcalculator_gabetest/__init__.py__author__ = 'Olivier Van Cutsem'
from electricitycostcalculator_testgabe import cost_calculator, oadr_signal, openei_tariff
__all__ = ["cost_calculator", "oadr_signal", "openei_tariff"]
PK ! rI 2 electricitycostcalculator_gabetest/__init__.py.bak__author__ = 'Olivier Van Cutsem'
from electricitycostcalculator import cost_calculator, oadr_signal, openei_tariff
__all__ = ["cost_calculator", "oadr_signal", "openei_tariff"]
PK ! 4 = electricitycostcalculator_gabetest/cost_calculator/.gitignore*.pyc
PK ! Lz=! ! > electricitycostcalculator_gabetest/cost_calculator/__init__.py__author__ = 'Olivier Van Cutsem'PK ! Lz=! ! B electricitycostcalculator_gabetest/cost_calculator/__init__.py.bak__author__ = 'Olivier Van Cutsem'PK ! ;h?M ?M E electricitycostcalculator_gabetest/cost_calculator/cost_calculator.py__author__ = 'Olivier Van Cutsem'
from .rate_structure import *
from .tariff_structure import TariffType
from dateutil.relativedelta import relativedelta
import pandas as pd
import pytz
class CostCalculator(object):
"""
This class is used to manipulate the building electricity cost:
- Bill calculation given a Smart Meter energy timeseries
- Electricity price timeseries between two dates
- Cost coefficients over a given period, for a linear optimization problem
- Metrics related to the tariff maximum demand
The main component of this class is called the "tariff_structure".
It is a dictionnary that lists electricity cost information for each type (fix, energy or demand).
The list for each type of billing stores "blocks" of data, that collect data about the tariffication of the electricity for a specific PERIOD of time.
Any time a new tariffication (e.g. a PDP event, or a new base tariff) must be added for a new time period, one just need to add the "block" of data in the corresponding list.
"""
# This default structure lists the tariffs type for most of the utilities in US
DEFAULT_TARIFF_MAP = {str(TariffType.FIX_CUSTOM_CHARGE.value): ChargeType.FIXED,
str(TariffType.ENERGY_CUSTOM_CHARGE.value): ChargeType.ENERGY,
str(TariffType.DEMAND_CUSTOM_CHARGE_SEASON.value): ChargeType.DEMAND,
str(TariffType.DEMAND_CUSTOM_CHARGE_TOU.value): ChargeType.DEMAND,
str(TariffType.PDP_ENERGY_CHARGE.value): ChargeType.ENERGY,
str(TariffType.PDP_ENERGY_CREDIT.value): ChargeType.ENERGY,
str(TariffType.PDP_DEMAND_CREDIT.value): ChargeType.DEMAND,
}
def __init__(self, type_tariffs_map=None):
"""
Initialize the class instance
:param type_tariffs_map: [optional] a dictionary that map the main type of tariffs used to describe the whole
billing logic to their type. DEFAULT_TARIFF_TYPE_LIST is used if type_tariffs_list is not specified.
Note: the method 'add_tariff' is used to build the core "tariff_structure" object structure.
"""
# This is the main structure, listing all the "tariff blocks" making up the whole tariff logic
self.__tariffstructures = {}
if type_tariffs_map is None: # The "basic" tariff types as the default ones
self.type_tariffs_map = self.DEFAULT_TARIFF_MAP
else:
self.type_tariffs_map = type_tariffs_map
# Initialize the list of "tariff blocks"
for label, type_tariff in list(self.type_tariffs_map.items()):
self.__tariffstructures[label] = self.generate_type_tariff(type_tariff)
# Useful data about the tariff
self.tariff_min_kw = 0 # The minimum peak demand to stay in this tariff
self.tariff_max_kw = float('inf') # The maximum peak demand to stay in this tariff
self.tariff_min_kwh = 0 # The minimum energy demand to stay in this tariff
self.tariff_max_kwh = float('inf') # The maximum energy demand to stay in this tariff
# --- Useful methods
def compute_bill(self, df, column_data=None, monthly_detailed=False):
"""
#TODO: create a class for the bill !
Return the bill corresponding to the electricity data in a data frame:
{
"label1": cost_detail_1
"label2": cost_detail_2
...
}
where:
- keys label_i corresponds to a type of tariff in the Enum TariffType and the values
- values cost_detail_i has one of the following form:
- if ENERGY or FIX tariff: cost_detail_i = (metric, cost) where metric is either the total energy or the period
- if DEMAND: cost_detail_i is dict where the keys are the price per kW and the values are tuples: (period-mask, max-power-value, max-power-date)
if monthly_detailed is set to True, the bill is detailed for each month:
{
"YY-MM":
{
"label1": (int or float, float) or a dict, -> the metric associated to the label1 and the corresponding cost (in $) in the month
"label2": (int or float, float) or a dict, -> the metric associated to the label2 and the corresponding cost (in $) in the month
...
}
}
:param df: a pandas dataframe containing energy consumption (in Wh) in the column 'column_data'.
If column data is None, it is assumed that only 1 column makes up the df
:param column_data: [optional] the label of the column containing the energy consumption values
:param monthly_detailed: [optional] if False, it is assumed that the df contains values for ONE billing period.
if True, the bill is detailed for each month of the calendar. Set to False by default.
:return: a dictionary representing the bill as described above
"""
ret = {}
# Initialize the returned structure
t_s = df.index[0]
t_i = datetime(year=t_s.year, month=t_s.month, day=1, tzinfo=t_s.tzinfo)
while t_i <= df.index[-1]:
ret[t_i.strftime("%Y-%m")] = {}
for k in list(self.__tariffstructures.keys()):
if self.type_tariffs_map[k] == ChargeType.DEMAND:
ret[t_i.strftime("%Y-%m")][k] = {} # a dict of price -> (max, cost)
else:
ret[t_i.strftime("%Y-%m")][k] = (0, 0) # a tuple
t_i += relativedelta(months=+1)
# Compute the bill for each of the tariff type, for each month
for label, tariff_data in list(self.__tariffstructures.items()):
l_blocks = self.get_tariff_struct(label, (df.index[0], df.index[-1])) # get all the tariff blocks for this period and this tariff type
for tariff_block in l_blocks:
tariff_cost_list = tariff_block.compute_bill(df, column_data) # this returns a dict of time-period pointing to tuple that contains both the metric of the bill and the cost
for time_label, bill_data in list(tariff_cost_list.items()):
self.update_bill_structure(ret[time_label], label, bill_data)
if monthly_detailed is False: # Aggregate all the months
return self.aggregate_monthly_bill(ret)
else:
return ret
def get_electricity_price(self, range_date, timestep):
"""
This function creates the electricity price signal for the specified time frame 'range_date', sampled at 'timestep'
period. It returns a pandas dataframes where the columns point to each type of tariff, specified in the argument
'type_tariffs_map' of the constructor.
:param range_date: a tuple (t_start, t_end) of type 'datetime', representing the period
:param timestep: an element of TariffElemPeriod enumeration (1h, 30min or 15min), representing the sampling
period
:return: a tuple (pd_prices, map_prices) containing:
- pd_prices: a pandas dataframe whose index is a datetime index and containing as many cols as there are
type_tariffs_map elements, i.e. the same keys as in __tariffstructures
- map_prices: a mapping between the cols label and the type of tariff (fix, energy or demand), being of type 'ChargeType'
"""
# Prepare the Pandas dataframe
(start_date_price, end_date_price) = range_date
date_list = pd.date_range(start=start_date_price, end=end_date_price, freq=str(timestep.value))
# Populate the dataframe for each label, for each period
ret_df = None
for label_tariff in list(self.__tariffstructures.keys()):
if self.type_tariffs_map[label_tariff] == ChargeType.FIXED: # fixed charges not in the elec price signal
continue
df_for_label = self.get_price_in_range(label_tariff, range_date, timestep)
if ret_df is None:
ret_df = df_for_label
else:
ret_df = pd.concat([ret_df, df_for_label], axis=1)
return ret_df, self.type_tariffs_map
def get_price_in_range(self, label_tariff, date_range, timestep):
"""
Generate a dataframe of the price of
remark: doesn't work with timestep > 1h ..
"""
# Prepare the Pandas dataframe
(start_date_price, end_date_price) = date_range
date_range = pd.date_range(start=start_date_price, end=end_date_price, freq=str(timestep.value))
ret_df = pd.DataFrame(index=date_range, columns=[label_tariff])
# # Select the corresponding blocks for each day and generate the time dataframe
for idx_day, df_day in ret_df.groupby(ret_df.index.date):
date_range_period = pd.date_range(start=df_day.index[0], periods=2, freq=str(timestep.value))
tariff_block = self.get_tariff_struct(label_tariff, (date_range_period[0], date_range_period[1]))
if len(tariff_block) > 0:
daily_rate = tariff_block[0].rate_schedule.get_daily_rate(df_day.index[0])
rate_df = tariff_block[0].get_daily_price_dataframe(daily_rate, df_day)
ret_df.loc[df_day.index[:], label_tariff] = rate_df['price'].values
return ret_df
def print_aggregated_bill(self, bill_struct, verbose=True):
"""
This method helps manipulating the bill returned by compute_bill().
It takes the bill as an argument and return a tuple (t, tt, ttt):
- t is the total cost
- tt is the total cost per type of tariff (energy, fix, demand)
- ttt is the cost for each tariff label
:param bill_struct: the dictionary returned by compute_bill()
:param verbose: [optional, default is True] print details
:return:
"""
monthly_detailed = False
# If the first keys of the dict point to smth that is not the tariff type, this is a monthly bill
first_keys_bill_struct = list(bill_struct.keys())
if first_keys_bill_struct[0] not in list(self.__tariffstructures.keys()):
monthly_detailed = True
if monthly_detailed is True: # This supposes the bill is calculated per natural month of the calendar
# Aggregation of all the months
acc_tot = 0.0
acc_per_chargetype = {ChargeType.FIXED: 0.0, ChargeType.ENERGY: 0.0, ChargeType.DEMAND: 0.0}
acc_per_label = {}
for k in list(self.type_tariffs_map.keys()):
acc_per_label[k] = 0.0
for m_key, bill_per_label in list(bill_struct.items()):
for lab_tariff, data in list(bill_per_label.items()):
if self.type_tariffs_map[lab_tariff] is not ChargeType.DEMAND:
cost_per_tariff = data[1]
else:
cost_per_tariff = 0.0
for p, data_demand in list(data.items()):
cost_per_tariff += p * data_demand['max-demand']
acc_tot += cost_per_tariff # second item in data is in dollar
acc_per_chargetype[self.type_tariffs_map[lab_tariff]] += cost_per_tariff
acc_per_label[lab_tariff] += cost_per_tariff
else:
# The bill is already aggregated for all the months
acc_tot = 0.0
acc_per_chargetype = {ChargeType.FIXED: 0.0, ChargeType.ENERGY: 0.0, ChargeType.DEMAND: 0.0}
for lab_tariff, data in list(bill_struct.items()):
if self.type_tariffs_map[lab_tariff] is not ChargeType.DEMAND:
cost_per_tariff = data[1]
else:
cost_per_tariff = 0.0
for p, data_demand in list(data.items()):
cost_per_tariff += p * data_demand['max-demand']
acc_tot += cost_per_tariff # second item in data is in dollar
acc_per_chargetype[self.type_tariffs_map[lab_tariff]] += cost_per_tariff
acc_per_label = bill_struct
if verbose:
# Total
print(("\n| Aggregated bill: {0} ($)".format(acc_tot)))
# Per type
print("\n| Total bill per type of charge:")
for t_key, v in list(acc_per_chargetype.items()):
print((" - Charge type '{0}': {1} ($)".format(str(t_key.value), v)))
# Per label
print("\n| Total bill per type or tariff:")
for l_key, v in list(acc_per_label.items()):
# TODO: print nicely the details ...
print((" - Type '{0}': {1} ($)".format(str(l_key), v)))
return acc_tot, acc_per_chargetype, acc_per_label
# --- Construction and internal methods
def add_tariff(self, tariff_obj, tariff_label, tariff_type=None):
"""
Add a tariff block structure that fell into the category "type_rate"
:param tariff_obj: a TariffBase (or children) object
:param tariff_label: the label of the tariff, in the keys given to the constructor
:param tariff_type: the type of tariff, an enum of ChargeType
:return: /
"""
# The tariff type (fix, demand or energy) is not specified: get it from the default structure
if tariff_type is None:
tariff_type = tariff_label
if tariff_label in list(self.DEFAULT_TARIFF_MAP.keys()):
tariff_type = self.DEFAULT_TARIFF_MAP[tariff_label]
else:
print(("[in add_tariff] Couldn't add the tariff object:" \
"The tariff_type is missing and couldn't be retrieved from the label '{0}'".format(tariff_label))) # debug
return
# The label tariff is a new one:
if tariff_label not in list(self.__tariffstructures.keys()):
self.__tariffstructures[tariff_label] = self.generate_type_tariff(tariff_type)
self.__tariffstructures[tariff_label]['list_blocks'].append(tariff_obj)
def get_tariff_struct(self, label_tariff, dates=None):
"""
Get the list of "tariff blocks" that influence the bill for the type of tariff "type_rate".
If "dates" is specified, only the blocks that are effective for that period are returned
:param label_tariff: a string pointing to the type of tariff
:param dates:[optional] a tuple of type datetime defining the period of selection
:return: a list of TariffBase (or children) describing the tariffs
"""
list_struct = self.__tariffstructures[label_tariff]['list_blocks']
if dates is None:
return list_struct
else:
(start_sel, end_sel) = dates
if start_sel.tzinfo is None and len(list_struct) > 0:
first_block = list_struct[0]
start_sel = start_sel.replace(tzinfo=pytz.timezone('UTC'))
if end_sel.tzinfo is None and len(list_struct) > 0:
first_block = list_struct[0]
end_sel = end_sel.replace(tzinfo=pytz.timezone('UTC'))
return [obj for obj in list_struct if ((obj.startdate <= start_sel <= obj.enddate) or (start_sel <= obj.startdate <= end_sel))]
def update_bill_structure(self, intermediate_monthly_bill, label_tariff, new_data):
"""
This method update the current monthly bill with new data for the same month:
- In case of "demand charge per (k)W", apply MAX
- In case of "energy charge per (k)Wh or fixed cost per month", apply SUM
:param intermediate_monthly_bill: the dict structure as return by the compute_bill() method, for a specific month key
:param label_tariff: a string indicating the tariff. Must be a key of self.__tariffstructures
:param new_data: a tuple (metric, cost) where:
- metric is either a float or an int, referring to the metric that influences the cost
- cost is a float, referring to the cost in $
:return:
"""
type_of_tariff = self.__tariffstructures[label_tariff]['type']
if type_of_tariff == ChargeType.DEMAND: # Demand: apply MAX
for p in list(new_data.keys()): # For each price -> dict (mask, max-p, date-max-p)
this_mask = new_data[p]['mask'] # get the new data mask
existing_mask_price = [k for k, v in list(intermediate_monthly_bill[label_tariff].items()) if v['mask'] == this_mask]
if len(existing_mask_price) > 0: # this mask has already been seen: APPLY MAX
existing_mask_price = existing_mask_price[0]
if new_data[p]['max-demand'] > intermediate_monthly_bill[label_tariff][existing_mask_price]['max-demand']:
#print("Demand rate update: for mask {0}, {0} is greater than {2}".format(this_mask, new_data[p]['max-demand'], intermediate_monthly_bill[label_tariff][existing_mask_price]['max-demand'])) # debug
del intermediate_monthly_bill[label_tariff][existing_mask_price]
intermediate_monthly_bill[label_tariff][p] = new_data[p]
else: # This is the first time this mask has been seen: store it
intermediate_monthly_bill[label_tariff][p] = new_data[p]
else: # energy or fixed cost: apply SUM
intermediate_monthly_bill[label_tariff] = (intermediate_monthly_bill[label_tariff][0] + new_data[0],
intermediate_monthly_bill[label_tariff][1] + new_data[1])
def aggregate_monthly_bill(self, monthly_bill):
"""
:param monthly_bill:
:return: /
"""
data_merge = None
for m, data_per_label in list(monthly_bill.items()):
if data_merge is None:
data_merge = data_per_label
else:
for label_tariff, data_tariff in list(data_per_label.items()):
if self.type_tariffs_map[label_tariff] == ChargeType.DEMAND: # take max
for p, data in list(data_tariff.items()): # For each price -> dict (mask, max, date)
this_mask = data['mask'] # get the new data mask
existing_mask_price = [k for k, v in list(data_merge[label_tariff].items()) if v['mask'] == this_mask]
if len(existing_mask_price) > 0: # this mask has already been seen: APPLY MAX
existing_mask_price = existing_mask_price[0]
if data['max-demand'] > data_merge[label_tariff][existing_mask_price]['max-demand']:
#print("Demand rate update: for mask {1}, {0} is greater than {2}".format(this_mask, data, data_merge[label_tariff][p])) # debug
del data_merge[label_tariff][existing_mask_price]
data_merge[label_tariff][p] = data
else: # This is the first time this price has been seen: store it
data_merge[label_tariff][p] = data
else: # sum
data_merge[label_tariff] = (data_merge[label_tariff][0] + data_tariff[0],
data_merge[label_tariff][1] + data_tariff[1])
return data_merge
@staticmethod
def generate_type_tariff(type_tariff):
return {'type': type_tariff,
'list_blocks': []}
PK ! +xGL L I electricitycostcalculator_gabetest/cost_calculator/cost_calculator.py.bak__author__ = 'Olivier Van Cutsem'
from .rate_structure import *
from .tariff_structure import TariffType
from dateutil.relativedelta import relativedelta
import pandas as pd
import pytz
class CostCalculator(object):
"""
This class is used to manipulate the building electricity cost:
- Bill calculation given a Smart Meter energy timeseries
- Electricity price timeseries between two dates
- Cost coefficients over a given period, for a linear optimization problem
- Metrics related to the tariff maximum demand
The main component of this class is called the "tariff_structure".
It is a dictionnary that lists electricity cost information for each type (fix, energy or demand).
The list for each type of billing stores "blocks" of data, that collect data about the tariffication of the electricity for a specific PERIOD of time.
Any time a new tariffication (e.g. a PDP event, or a new base tariff) must be added for a new time period, one just need to add the "block" of data in the corresponding list.
"""
# This default structure lists the tariffs type for most of the utilities in US
DEFAULT_TARIFF_MAP = {str(TariffType.FIX_CUSTOM_CHARGE.value): ChargeType.FIXED,
str(TariffType.ENERGY_CUSTOM_CHARGE.value): ChargeType.ENERGY,
str(TariffType.DEMAND_CUSTOM_CHARGE_SEASON.value): ChargeType.DEMAND,
str(TariffType.DEMAND_CUSTOM_CHARGE_TOU.value): ChargeType.DEMAND,
str(TariffType.PDP_ENERGY_CHARGE.value): ChargeType.ENERGY,
str(TariffType.PDP_ENERGY_CREDIT.value): ChargeType.ENERGY,
str(TariffType.PDP_DEMAND_CREDIT.value): ChargeType.DEMAND,
}
def __init__(self, type_tariffs_map=None):
"""
Initialize the class instance
:param type_tariffs_map: [optional] a dictionary that map the main type of tariffs used to describe the whole
billing logic to their type. DEFAULT_TARIFF_TYPE_LIST is used if type_tariffs_list is not specified.
Note: the method 'add_tariff' is used to build the core "tariff_structure" object structure.
"""
# This is the main structure, listing all the "tariff blocks" making up the whole tariff logic
self.__tariffstructures = {}
if type_tariffs_map is None: # The "basic" tariff types as the default ones
self.type_tariffs_map = self.DEFAULT_TARIFF_MAP
else:
self.type_tariffs_map = type_tariffs_map
# Initialize the list of "tariff blocks"
for label, type_tariff in self.type_tariffs_map.items():
self.__tariffstructures[label] = self.generate_type_tariff(type_tariff)
# Useful data about the tariff
self.tariff_min_kw = 0 # The minimum peak demand to stay in this tariff
self.tariff_max_kw = float('inf') # The maximum peak demand to stay in this tariff
self.tariff_min_kwh = 0 # The minimum energy demand to stay in this tariff
self.tariff_max_kwh = float('inf') # The maximum energy demand to stay in this tariff
# --- Useful methods
def compute_bill(self, df, column_data=None, monthly_detailed=False):
"""
#TODO: create a class for the bill !
Return the bill corresponding to the electricity data in a data frame:
{
"label1": cost_detail_1
"label2": cost_detail_2
...
}
where:
- keys label_i corresponds to a type of tariff in the Enum TariffType and the values
- values cost_detail_i has one of the following form:
- if ENERGY or FIX tariff: cost_detail_i = (metric, cost) where metric is either the total energy or the period
- if DEMAND: cost_detail_i is dict where the keys are the price per kW and the values are tuples: (period-mask, max-power-value, max-power-date)
if monthly_detailed is set to True, the bill is detailed for each month:
{
"YY-MM":
{
"label1": (int or float, float) or a dict, -> the metric associated to the label1 and the corresponding cost (in $) in the month
"label2": (int or float, float) or a dict, -> the metric associated to the label2 and the corresponding cost (in $) in the month
...
}
}
:param df: a pandas dataframe containing energy consumption (in Wh) in the column 'column_data'.
If column data is None, it is assumed that only 1 column makes up the df
:param column_data: [optional] the label of the column containing the energy consumption values
:param monthly_detailed: [optional] if False, it is assumed that the df contains values for ONE billing period.
if True, the bill is detailed for each month of the calendar. Set to False by default.
:return: a dictionary representing the bill as described above
"""
ret = {}
# Initialize the returned structure
t_s = df.index[0]
t_i = datetime(year=t_s.year, month=t_s.month, day=1, tzinfo=t_s.tzinfo)
while t_i <= df.index[-1]:
ret[t_i.strftime("%Y-%m")] = {}
for k in self.__tariffstructures.keys():
if self.type_tariffs_map[k] == ChargeType.DEMAND:
ret[t_i.strftime("%Y-%m")][k] = {} # a dict of price -> (max, cost)
else:
ret[t_i.strftime("%Y-%m")][k] = (0, 0) # a tuple
t_i += relativedelta(months=+1)
# Compute the bill for each of the tariff type, for each month
for label, tariff_data in self.__tariffstructures.items():
l_blocks = self.get_tariff_struct(label, (df.index[0], df.index[-1])) # get all the tariff blocks for this period and this tariff type
for tariff_block in l_blocks:
tariff_cost_list = tariff_block.compute_bill(df, column_data) # this returns a dict of time-period pointing to tuple that contains both the metric of the bill and the cost
for time_label, bill_data in tariff_cost_list.items():
self.update_bill_structure(ret[time_label], label, bill_data)
if monthly_detailed is False: # Aggregate all the months
return self.aggregate_monthly_bill(ret)
else:
return ret
def get_electricity_price(self, range_date, timestep):
"""
This function creates the electricity price signal for the specified time frame 'range_date', sampled at 'timestep'
period. It returns a pandas dataframes where the columns point to each type of tariff, specified in the argument
'type_tariffs_map' of the constructor.
:param range_date: a tuple (t_start, t_end) of type 'datetime', representing the period
:param timestep: an element of TariffElemPeriod enumeration (1h, 30min or 15min), representing the sampling
period
:return: a tuple (pd_prices, map_prices) containing:
- pd_prices: a pandas dataframe whose index is a datetime index and containing as many cols as there are
type_tariffs_map elements, i.e. the same keys as in __tariffstructures
- map_prices: a mapping between the cols label and the type of tariff (fix, energy or demand), being of type 'ChargeType'
"""
# Prepare the Pandas dataframe
(start_date_price, end_date_price) = range_date
date_list = pd.date_range(start=start_date_price, end=end_date_price, freq=str(timestep.value))
# Populate the dataframe for each label, for each period
ret_df = None
for label_tariff in self.__tariffstructures.keys():
if self.type_tariffs_map[label_tariff] == ChargeType.FIXED: # fixed charges not in the elec price signal
continue
df_for_label = self.get_price_in_range(label_tariff, range_date, timestep)
if ret_df is None:
ret_df = df_for_label
else:
ret_df = pd.concat([ret_df, df_for_label], axis=1)
return ret_df, self.type_tariffs_map
def get_price_in_range(self, label_tariff, date_range, timestep):
"""
Generate a dataframe of the price of
remark: doesn't work with timestep > 1h ..
"""
# Prepare the Pandas dataframe
(start_date_price, end_date_price) = date_range
date_range = pd.date_range(start=start_date_price, end=end_date_price, freq=str(timestep.value))
ret_df = pd.DataFrame(index=date_range, columns=[label_tariff])
# # Select the corresponding blocks for each day and generate the time dataframe
for idx_day, df_day in ret_df.groupby(ret_df.index.date):
date_range_period = pd.date_range(start=df_day.index[0], periods=2, freq=str(timestep.value))
tariff_block = self.get_tariff_struct(label_tariff, (date_range_period[0], date_range_period[1]))
if len(tariff_block) > 0:
daily_rate = tariff_block[0].rate_schedule.get_daily_rate(df_day.index[0])
rate_df = tariff_block[0].get_daily_price_dataframe(daily_rate, df_day)
ret_df.loc[df_day.index[:], label_tariff] = rate_df['price'].values
return ret_df
def print_aggregated_bill(self, bill_struct, verbose=True):
"""
This method helps manipulating the bill returned by compute_bill().
It takes the bill as an argument and return a tuple (t, tt, ttt):
- t is the total cost
- tt is the total cost per type of tariff (energy, fix, demand)
- ttt is the cost for each tariff label
:param bill_struct: the dictionary returned by compute_bill()
:param verbose: [optional, default is True] print details
:return:
"""
monthly_detailed = False
# If the first keys of the dict point to smth that is not the tariff type, this is a monthly bill
first_keys_bill_struct = list(bill_struct.keys())
if first_keys_bill_struct[0] not in self.__tariffstructures.keys():
monthly_detailed = True
if monthly_detailed is True: # This supposes the bill is calculated per natural month of the calendar
# Aggregation of all the months
acc_tot = 0.0
acc_per_chargetype = {ChargeType.FIXED: 0.0, ChargeType.ENERGY: 0.0, ChargeType.DEMAND: 0.0}
acc_per_label = {}
for k in self.type_tariffs_map.keys():
acc_per_label[k] = 0.0
for m_key, bill_per_label in bill_struct.items():
for lab_tariff, data in bill_per_label.items():
if self.type_tariffs_map[lab_tariff] is not ChargeType.DEMAND:
cost_per_tariff = data[1]
else:
cost_per_tariff = 0.0
for p, data_demand in data.items():
cost_per_tariff += p * data_demand['max-demand']
acc_tot += cost_per_tariff # second item in data is in dollar
acc_per_chargetype[self.type_tariffs_map[lab_tariff]] += cost_per_tariff
acc_per_label[lab_tariff] += cost_per_tariff
else:
# The bill is already aggregated for all the months
acc_tot = 0.0
acc_per_chargetype = {ChargeType.FIXED: 0.0, ChargeType.ENERGY: 0.0, ChargeType.DEMAND: 0.0}
for lab_tariff, data in bill_struct.items():
if self.type_tariffs_map[lab_tariff] is not ChargeType.DEMAND:
cost_per_tariff = data[1]
else:
cost_per_tariff = 0.0
for p, data_demand in data.items():
cost_per_tariff += p * data_demand['max-demand']
acc_tot += cost_per_tariff # second item in data is in dollar
acc_per_chargetype[self.type_tariffs_map[lab_tariff]] += cost_per_tariff
acc_per_label = bill_struct
if verbose:
# Total
print("\n| Aggregated bill: {0} ($)".format(acc_tot))
# Per type
print("\n| Total bill per type of charge:")
for t_key, v in acc_per_chargetype.items():
print(" - Charge type '{0}': {1} ($)".format(str(t_key.value), v))
# Per label
print("\n| Total bill per type or tariff:")
for l_key, v in acc_per_label.items():
# TODO: print nicely the details ...
print(" - Type '{0}': {1} ($)".format(str(l_key), v))
return acc_tot, acc_per_chargetype, acc_per_label
# --- Construction and internal methods
def add_tariff(self, tariff_obj, tariff_label, tariff_type=None):
"""
Add a tariff block structure that fell into the category "type_rate"
:param tariff_obj: a TariffBase (or children) object
:param tariff_label: the label of the tariff, in the keys given to the constructor
:param tariff_type: the type of tariff, an enum of ChargeType
:return: /
"""
# The tariff type (fix, demand or energy) is not specified: get it from the default structure
if tariff_type is None:
tariff_type = tariff_label
if tariff_label in self.DEFAULT_TARIFF_MAP.keys():
tariff_type = self.DEFAULT_TARIFF_MAP[tariff_label]
else:
print("[in add_tariff] Couldn't add the tariff object:" \
"The tariff_type is missing and couldn't be retrieved from the label '{0}'".format(tariff_label)) # debug
return
# The label tariff is a new one:
if tariff_label not in self.__tariffstructures.keys():
self.__tariffstructures[tariff_label] = self.generate_type_tariff(tariff_type)
self.__tariffstructures[tariff_label]['list_blocks'].append(tariff_obj)
def get_tariff_struct(self, label_tariff, dates=None):
"""
Get the list of "tariff blocks" that influence the bill for the type of tariff "type_rate".
If "dates" is specified, only the blocks that are effective for that period are returned
:param label_tariff: a string pointing to the type of tariff
:param dates:[optional] a tuple of type datetime defining the period of selection
:return: a list of TariffBase (or children) describing the tariffs
"""
list_struct = self.__tariffstructures[label_tariff]['list_blocks']
if dates is None:
return list_struct
else:
(start_sel, end_sel) = dates
if start_sel.tzinfo is None and len(list_struct) > 0:
first_block = list_struct[0]
start_sel = start_sel.replace(tzinfo=pytz.timezone('UTC'))
if end_sel.tzinfo is None and len(list_struct) > 0:
first_block = list_struct[0]
end_sel = end_sel.replace(tzinfo=pytz.timezone('UTC'))
return [obj for obj in list_struct if ((obj.startdate <= start_sel <= obj.enddate) or (start_sel <= obj.startdate <= end_sel))]
def update_bill_structure(self, intermediate_monthly_bill, label_tariff, new_data):
"""
This method update the current monthly bill with new data for the same month:
- In case of "demand charge per (k)W", apply MAX
- In case of "energy charge per (k)Wh or fixed cost per month", apply SUM
:param intermediate_monthly_bill: the dict structure as return by the compute_bill() method, for a specific month key
:param label_tariff: a string indicating the tariff. Must be a key of self.__tariffstructures
:param new_data: a tuple (metric, cost) where:
- metric is either a float or an int, referring to the metric that influences the cost
- cost is a float, referring to the cost in $
:return:
"""
type_of_tariff = self.__tariffstructures[label_tariff]['type']
if type_of_tariff == ChargeType.DEMAND: # Demand: apply MAX
for p in new_data.keys(): # For each price -> dict (mask, max-p, date-max-p)
this_mask = new_data[p]['mask'] # get the new data mask
existing_mask_price = [k for k, v in intermediate_monthly_bill[label_tariff].items() if v['mask'] == this_mask]
if len(existing_mask_price) > 0: # this mask has already been seen: APPLY MAX
existing_mask_price = existing_mask_price[0]
if new_data[p]['max-demand'] > intermediate_monthly_bill[label_tariff][existing_mask_price]['max-demand']:
#print("Demand rate update: for mask {0}, {0} is greater than {2}".format(this_mask, new_data[p]['max-demand'], intermediate_monthly_bill[label_tariff][existing_mask_price]['max-demand'])) # debug
del intermediate_monthly_bill[label_tariff][existing_mask_price]
intermediate_monthly_bill[label_tariff][p] = new_data[p]
else: # This is the first time this mask has been seen: store it
intermediate_monthly_bill[label_tariff][p] = new_data[p]
else: # energy or fixed cost: apply SUM
intermediate_monthly_bill[label_tariff] = (intermediate_monthly_bill[label_tariff][0] + new_data[0],
intermediate_monthly_bill[label_tariff][1] + new_data[1])
def aggregate_monthly_bill(self, monthly_bill):
"""
:param monthly_bill:
:return: /
"""
data_merge = None
for m, data_per_label in monthly_bill.items():
if data_merge is None:
data_merge = data_per_label
else:
for label_tariff, data_tariff in data_per_label.items():
if self.type_tariffs_map[label_tariff] == ChargeType.DEMAND: # take max
for p, data in data_tariff.items(): # For each price -> dict (mask, max, date)
this_mask = data['mask'] # get the new data mask
existing_mask_price = [k for k, v in data_merge[label_tariff].items() if v['mask'] == this_mask]
if len(existing_mask_price) > 0: # this mask has already been seen: APPLY MAX
existing_mask_price = existing_mask_price[0]
if data['max-demand'] > data_merge[label_tariff][existing_mask_price]['max-demand']:
#print("Demand rate update: for mask {1}, {0} is greater than {2}".format(this_mask, data, data_merge[label_tariff][p])) # debug
del data_merge[label_tariff][existing_mask_price]
data_merge[label_tariff][p] = data
else: # This is the first time this price has been seen: store it
data_merge[label_tariff][p] = data
else: # sum
data_merge[label_tariff] = (data_merge[label_tariff][0] + data_tariff[0],
data_merge[label_tariff][1] + data_tariff[1])
return data_merge
@staticmethod
def generate_type_tariff(type_tariff):
return {'type': type_tariff,
'list_blocks': []}
PK ! 9K K D electricitycostcalculator_gabetest/cost_calculator/rate_structure.py__author__ = 'Olivier Van Cutsem'
from enum import Enum
from datetime import datetime
import holidays
# --------------- Schedule structures --------------- #
class ChargeType(Enum):
FIXED = 'fix',
DEMAND = 'demand',
ENERGY = 'energy',
class TouRateSchedule:
"""
This structure stores the Time-Of-Use rates related to power or energy.
It is made of a set of methods that manipulate a dict formatted as follow:
{
"monthly_label1":
{
"months_list": [m1, m2, ...],
"daily_rates":
{
"daily_label1:
{
"days_list": [d1, d2, ...],
"rates": list OR float
},
...
}
},
...
}
Remark: a month spans from 1 (january) to 12 (december) ; a day spans from 0 (sunday) to 6 (saturday)
"""
# TODO: use BlockRate instead of assuming it's a float !
# Keys used internally
MONTHLIST_KEY = 'months_list'
DAILY_RATE_KEY = 'daily_rates'
DAYSLIST_KEY = 'days_list'
RATES_KEY = 'rates'
def __init__(self, rates_schedule):
"""
Constructor
:param rates_schedule: a dict formatted as explain in the class description
"""
# TODO: assert the format is correct
self.__rates = rates_schedule
def get_from_timestamp(self, date):
"""
Return the rate corresponding to a given timestamp
:param date: a float, the timestamp
:return: a float, the rate corresponding to the timestamp. None if there is no associated rate
"""
# Get (m, d, h, m) from date
if type(date) is float or type(date) is int:
date_struct = datetime.fromtimestamp(date)
else:
date_struct = date
m_date = date_struct.month
d_date = date_struct.weekday()
h_date = date_struct.hour
min_date = date_struct.minute
rates = self.get_rate(m_date, d_date)
return self.get_rate_in_day(rates, (h_date, min_date))
def get_daily_rate(self, date):
"""
Return the daily rates, as a vector sampled at a given period
:param date: a float, the timestamp
:return: a list of float
"""
if type(date) is float or type(date) is int:
date_struct = datetime.fromtimestamp(date)
else:
date_struct = date
m_date = date_struct.month
d_date = self.get_day_in_the_week(date_struct)
rate_struct = self.get_rate(m_date, d_date)
if type(rate_struct) is not list: # hourly flat rate
return [rate_struct]
else:
return rate_struct
# --- private
@staticmethod
def get_day_in_the_week(date_sel):
"""
TODO write description
:param date_sel:
:return:
"""
if date_sel in holidays.US(state='CA', years=date_sel.year):
return 0 # Hardcoded: holidays are like Sundays ...
else:
return date_sel.weekday()
def get_rate_in_day(self, rate_struct, time_select):
"""
Return the rate in 'rate_struct' corresponding to instant "time_select"
:param rate_struct: either a float or an int, representing the rate(s) of the day
:param time_select: a tuple (h, m) representing the hour and minute to select
:return: a float, the rate at the selected time
"""
(h, m) = time_select
if type(rate_struct) is not list:
return rate_struct
else:
idx = (h + m/60.0 ) * len(rate_struct) / 24.0
return rate_struct[int(idx)]
def get_rate(self, m_date, d_date):
"""
TODO write description
:param self:
:param m_date:
:param d_date:
:return:
"""
for m_lab, m_data in list(self.__rates.items()):
if m_date in m_data[self.MONTHLIST_KEY]:
for d_lab, d_data in list(m_data[self.DAILY_RATE_KEY].items()):
if d_date in d_data[self.DAYSLIST_KEY]:
return d_data[self.RATES_KEY]
@property
def periods_in_day(self):
"""
TODO write description
:return:
"""
# take a random day and check the vector length
random_day = datetime(2000, 1, 1, hour=0, minute=0, second=0) # the year doesn't matter
vector_data = self.get_daily_rate(random_day)
return len(vector_data)
@property
def main_structure(self):
"""
The raw tariff rates
"""
return self.__rates
class BlockRate:
"""
This class stores and manipulates the rate of energy that vary as a function of the total consumption energy
"""
def __init__(self, cost_base, block_rate=None):
self.__rates = [cost_base]
self.__thresholds = [0]
if block_rate is not None:
(costs, thres) = block_rate
self.__rates += costs
self.__thresholds += thres
self.__thresholds.append(float('inf'))
def get_rate(self, acc=None):
"""
:param acc:
:return:
"""
if acc is None:
return self.__rates[0]
else:
return [self.__rates[i] for i in range(len(self.__rates)) if self.__thresholds[i] <= acc < self.__thresholds[i+1]][0]
PK ! |U? ? H electricitycostcalculator_gabetest/cost_calculator/rate_structure.py.bak__author__ = 'Olivier Van Cutsem'
from enum import Enum
from datetime import datetime
import holidays
# --------------- Schedule structures --------------- #
class ChargeType(Enum):
FIXED = 'fix',
DEMAND = 'demand',
ENERGY = 'energy',
class TouRateSchedule:
"""
This structure stores the Time-Of-Use rates related to power or energy.
It is made of a set of methods that manipulate a dict formatted as follow:
{
"monthly_label1":
{
"months_list": [m1, m2, ...],
"daily_rates":
{
"daily_label1:
{
"days_list": [d1, d2, ...],
"rates": list OR float
},
...
}
},
...
}
Remark: a month spans from 1 (january) to 12 (december) ; a day spans from 0 (sunday) to 6 (saturday)
"""
# TODO: use BlockRate instead of assuming it's a float !
# Keys used internally
MONTHLIST_KEY = 'months_list'
DAILY_RATE_KEY = 'daily_rates'
DAYSLIST_KEY = 'days_list'
RATES_KEY = 'rates'
def __init__(self, rates_schedule):
"""
Constructor
:param rates_schedule: a dict formatted as explain in the class description
"""
# TODO: assert the format is correct
self.__rates = rates_schedule
def get_from_timestamp(self, date):
"""
Return the rate corresponding to a given timestamp
:param date: a float, the timestamp
:return: a float, the rate corresponding to the timestamp. None if there is no associated rate
"""
# Get (m, d, h, m) from date
if type(date) is float or type(date) is int:
date_struct = datetime.fromtimestamp(date)
else:
date_struct = date
m_date = date_struct.month
d_date = date_struct.weekday()
h_date = date_struct.hour
min_date = date_struct.minute
rates = self.get_rate(m_date, d_date)
return self.get_rate_in_day(rates, (h_date, min_date))
def get_daily_rate(self, date):
"""
Return the daily rates, as a vector sampled at a given period
:param date: a float, the timestamp
:return: a list of float
"""
if type(date) is float or type(date) is int:
date_struct = datetime.fromtimestamp(date)
else:
date_struct = date
m_date = date_struct.month
d_date = self.get_day_in_the_week(date_struct)
rate_struct = self.get_rate(m_date, d_date)
if type(rate_struct) is not list: # hourly flat rate
return [rate_struct]
else:
return rate_struct
# --- private
@staticmethod
def get_day_in_the_week(date_sel):
"""
TODO write description
:param date_sel:
:return:
"""
if date_sel in holidays.US(state='CA', years=date_sel.year):
return 0 # Hardcoded: holidays are like Sundays ...
else:
return date_sel.weekday()
def get_rate_in_day(self, rate_struct, time_select):
"""
Return the rate in 'rate_struct' corresponding to instant "time_select"
:param rate_struct: either a float or an int, representing the rate(s) of the day
:param time_select: a tuple (h, m) representing the hour and minute to select
:return: a float, the rate at the selected time
"""
(h, m) = time_select
if type(rate_struct) is not list:
return rate_struct
else:
idx = (h + m/60.0 ) * len(rate_struct) / 24.0
return rate_struct[int(idx)]
def get_rate(self, m_date, d_date):
"""
TODO write description
:param self:
:param m_date:
:param d_date:
:return:
"""
for m_lab, m_data in self.__rates.items():
if m_date in m_data[self.MONTHLIST_KEY]:
for d_lab, d_data in m_data[self.DAILY_RATE_KEY].items():
if d_date in d_data[self.DAYSLIST_KEY]:
return d_data[self.RATES_KEY]
@property
def periods_in_day(self):
"""
TODO write description
:return:
"""
# take a random day and check the vector length
random_day = datetime(2000, 1, 1, hour=0, minute=0, second=0) # the year doesn't matter
vector_data = self.get_daily_rate(random_day)
return len(vector_data)
@property
def main_structure(self):
"""
The raw tariff rates
"""
return self.__rates
class BlockRate:
"""
This class stores and manipulates the rate of energy that vary as a function of the total consumption energy
"""
def __init__(self, cost_base, block_rate=None):
self.__rates = [cost_base]
self.__thresholds = [0]
if block_rate is not None:
(costs, thres) = block_rate
self.__rates += costs
self.__thresholds += thres
self.__thresholds.append(float('inf'))
def get_rate(self, acc=None):
"""
:param acc:
:return:
"""
if acc is None:
return self.__rates[0]
else:
return [self.__rates[i] for i in range(len(self.__rates)) if self.__thresholds[i] <= acc < self.__thresholds[i+1]][0]
PK ! Ν": ": F electricitycostcalculator_gabetest/cost_calculator/tariff_structure.py__author__ = 'Olivier Van Cutsem'
from abc import abstractmethod
from enum import Enum
from datetime import datetime
import calendar
import pandas as pd
# --------------- TARIFF structures --------------- #
class TariffType(Enum):
FIX_CUSTOM_CHARGE = 'customer_fix_charge'
ENERGY_CUSTOM_CHARGE = 'customer_energy_charge'
DEMAND_CUSTOM_CHARGE_SEASON = 'customer_demand_charge_season'
DEMAND_CUSTOM_CHARGE_TOU = 'customer_demand_charge_tou'
PDP_ENERGY_CHARGE = 'pdp_event_energy_charge'
PDP_ENERGY_CREDIT = 'pdp_non_event_energy_credit'
PDP_DEMAND_CREDIT = 'pdp_non_event_demand_credit'
class TariffElemPeriod(Enum):
MONTHLY = 'M'
DAILY = 'D'
HOURLY = '1h'
HALFLY = '30min'
QUARTERLY = '15min'
class TariffElemMetricUnit(Enum):
EN_WH = 1
DEMAND_W = 1
EN_KWH = 1000.0
DEMAND_KW = 1000.0
class TariffElemCostUnit(Enum):
CENT = 0.01
DOLLAR = 1
class TariffBase(object):
"""
This abstract class represent the base of any tariffication structure.
The main components are the starting and ending date of the structure.
"""
def __init__(self, dates, unit_cost, name=None):
# Starting and ending dates, as timestamps
ts, te = dates
self.__startdate = ts
self.__enddate = te
self.name = name
self.unit_cost = unit_cost
def compute_bill(self, df, data_col=None):
"""
Compute the bill due to the power/energy consumption in df, for each billing period specified in billing_periods
It outputs a dictionary formatted as follow:
{
"bill_period_label1": (float or dict, float), -> the monthly 'metric' and its cost
...
}
:param df: a pandas dataframe containing power consumption timeseries
:param billing_periods: a dictionary mapping the billing periods label to a tuple (t_start, t_end) of datetime,
defining the period related to the billing label
:return: a dictionary formatted as in this method signature
"""
ret = {}
# Select only the data in this tariff window
start_sel = self.startdate
start_sel = start_sel.replace(tzinfo=df.index[0].tzinfo)
end_sel = self.enddate
end_sel = end_sel.replace(tzinfo=df.index[0].tzinfo)
mask = (df.index >= start_sel) & (df.index <= end_sel)
df = df.loc[mask]
# Loop over the months
t_s = df.index[0]
last_day_of_month = calendar.monthrange(t_s.year, t_s.month)[1] # The last day of this month
t_e = datetime(t_s.year, t_s.month, last_day_of_month, hour=23, minute=59, second=59, tzinfo=t_s.tzinfo) # end of the current month
t_e = min(df.index[-1], t_e)
while t_s <= t_e:
mask = (df.index >= t_s) & (df.index <= t_e)
df_month = df.loc[mask]
monthly_bill = self.compute_monthly_bill(df_month, data_col)
ret[t_s.strftime("%Y-%m")] = monthly_bill
# Prepare the next billing month
month = t_e.month + 1
year = t_e.year
if month >= 13:
month = 1
year += 1
t_s = datetime(year, month, 1, hour=0, minute=0, second=0, tzinfo=t_s.tzinfo)
last_day_of_month = calendar.monthrange(year, month)[1]
t_e = datetime(year, month, last_day_of_month, hour=23, minute=59, second=59, tzinfo=t_s.tzinfo)
t_e = min(df.index[-1], t_e)
return ret
@abstractmethod
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the monthly bill due to the power/energy consumption in df
:param df: a pandas dataframe
:param data_col: the column label containing the data
:return: a tuple (float, float) -> (value, cost), representing the bill and the corresponding metric linked to the cost
"""
pass
@property
def startdate(self):
"""
GETTER of the tariff starting date
:return: a timestamp
"""
return self.__startdate
@property
def enddate(self):
"""
GETTER of the tariff end date
:return: a timestamp
"""
return self.__enddate
@abstractmethod
def period_metric(self):
pass
@abstractmethod
def get_price_from_timestamp(self, timestamp):
pass
# --------------- FIXED TARIFF --------------- #
class FixedTariff(TariffBase):
"""
Represent a tariff fixed over a given period (among TariffPeriod)
"""
def __init__(self, dates, rate_value, unit_cost=TariffElemCostUnit.DOLLAR, bill_period=TariffElemPeriod.MONTHLY, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param bill_period: the period
:param name: see FixedTariff init
"""
super(FixedTariff, self).__init__(dates, unit_cost, name)
self.__rate_period = bill_period
self.__rate_value = rate_value
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the monthly bill due to a fixed periodic cost
:param df: a pandas dataframe
:return: a tuple (float, float), representing the bill and the duration (in months)
"""
first_day = df.index[0].day
last_day = df.index[-1].day
nb_days = last_day - first_day + 1
nb_days_per_month = 365/12
bill = 0
if self.__rate_period == TariffElemPeriod.MONTHLY:
bill = self.__rate_value * nb_days/nb_days_per_month # a fraction of the month
elif self.__rate_period == TariffElemPeriod.DAILY:
bill = self.__rate_value * nb_days # sum of each day
return nb_days, bill
def period_metric(self):
return self.__rate_period
def get_price_from_timestamp(self, timestamp):
return self.__rate_value
# --------------- TOU TARIFFs --------------- #
class TimeOfUseTariff(TariffBase):
"""
This class represents a tariff fixed over a given period (among TariffElemPeriod)
"""
def __init__(self, dates, rate_schedule, unit_metric, unit_cost, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param rate_list:
:param time_schedule: TODO
:param name: see FixedTariff init
"""
super(TimeOfUseTariff, self).__init__(dates, unit_cost, name)
self.__schedule = rate_schedule # A table mapping (month, day) to hourly rate index
self.__unit_metric = unit_metric
@abstractmethod
def compute_monthly_bill(self, df, data_col=None):
"""
idem super
"""
pass
@property
def rate_schedule(self):
return self.__schedule
@property
def unit_metric(self):
return self.__unit_metric
def period_metric(self):
nb_periods_in_day = self.__schedule.periods_in_day
# TODO: replace ifs by map
if nb_periods_in_day == 24:
return TariffElemPeriod.HOURLY
elif nb_periods_in_day == 24 * 2:
return TariffElemPeriod.HALFLY
elif nb_periods_in_day == 24 * 4:
return TariffElemPeriod.QUARTERLY
else:
return TariffElemPeriod.DAILY
def get_price_from_timestamp(self, timestamp):
return self.__schedule.get_from_timestamp(timestamp)
@staticmethod
def get_daily_price_dataframe(daily_rate, df_day):
# Constructing the dataframe for an easier manipulation of time
period_rate = len(daily_rate) / 24.0
# In some cases the day might not be full: missing data or DST
daily_prices = [daily_rate[int((df_day.index[i].hour + df_day.index[i].minute / 60.0) * period_rate)] for i in range(len(df_day.index))]
data = {'date': df_day.index[:], 'price': daily_prices}
df_prices = pd.DataFrame(data=data)
df_prices.set_index('date')
return df_prices
class TouDemandChargeTariff(TimeOfUseTariff):
"""
This class represents a Time Of Use Demand Charge tariff
"""
def __init__(self, dates, time_schedule, unit_metric=TariffElemMetricUnit.DEMAND_KW, unit_cost=TariffElemCostUnit.DOLLAR, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param rate_list: TODO
:param time_schedule: TODO
:param name: see FixedTariff init
"""
super(TouDemandChargeTariff, self).__init__(dates, time_schedule, unit_metric, unit_cost, name)
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the bill due to a TOU tariff
:param df: a pandas dataframe
:return: a tuple (dict, float) -> ({p1: (max_power_p1, time_max_p1), p2: (max_power_p2, time_max_p2), cost)
"""
# Scaling the power unit and cost
metric_unit_mult = float(self.unit_metric.value)
metric_price_mult = float(self.unit_cost.value)
max_per_set = {}
# df is in kWh and demand in kW: convert to Power
timestep_data = self.get_pd_timestep_data(df)
for idx, df_day in df.groupby(df.index.date):
daily_rate = self.rate_schedule.get_daily_rate(df_day.index[0])
set_of_daily_prices = set(daily_rate)
df_prices = self.get_daily_price_dataframe(daily_rate, df_day)
power_coeff = 1
if timestep_data == '15T':
power_coeff = 4
elif timestep_data == '30T':
power_coeff = 2
elif timestep_data == '60T' or timestep_data == 'H':
power_coeff = 1
for day_p in set_of_daily_prices:
# Create the mask in the day for this price
mask_price = df_prices['price'] == day_p
mask_price = mask_price.tolist()
date_max_period = None
if data_col is not None:
df_masked = df_day.loc[mask_price, data_col]
if len(df_masked) > 0:
date_max_period = df_masked.idxmax()
else:
df_masked = df_day.loc[mask_price]
if len(df_masked) > 0:
date_max_period = df_masked.idxmax()
if date_max_period is None:
continue
if data_col is not None:
max_power_period = df_day.loc[date_max_period, data_col] / metric_unit_mult
else:
max_power_period = df_day[date_max_period] / metric_unit_mult
max_power_period *= power_coeff # from kWh to kW
# Search for the same mask and update the value if a new mask
mask_price24h = mask_price
if len(mask_price) != len(daily_rate): # if this price is already in the list, take the corresponding mask ...
price_key = metric_price_mult * day_p
if price_key in list(max_per_set.keys()):
mask_price24h = max_per_set[price_key]['mask']
add_this_demand = True
existing_mask_price = [k for k, v in list(max_per_set.items()) if v['mask'] == mask_price24h]
if len(existing_mask_price) > 0: # Find the identical mask over the day
existing_mask_price = existing_mask_price[0]
if max_power_period > max_per_set[existing_mask_price]['max-demand']: # Check if the corresponding demand is greater
del max_per_set[existing_mask_price] # delete the former value, and add it (after)
else:
add_this_demand = False
# This is the first time this mask is seen OR this new demand is higher than the corresponding former: add it
if add_this_demand:
if type(date_max_period) is not pd.Timestamp:
max_power_date = date_max_period[data_col].to_pydatetime()
else:
max_power_date = date_max_period.to_pydatetime()
# The mask must be 24 hour long:
# it might not be the case for the first and last day, and the DST
price_key = metric_price_mult * day_p
max_per_set[price_key] = {'mask': mask_price24h, 'max-demand': max_power_period, 'max-demand-date': max_power_date}
return max_per_set
def get_pd_timestep_data(self, df):
"""
Return the most likely data frequency
:return:
"""
freq = 1
# Dataframe is complete
if df.index.freq is not None:
return df.index.freq
# Days are missing: loop through the day until getting a frequency
for idx, df_day in df.groupby(df.index.date):
if df_day.index.freq is not None:
return df_day.index.freq
elif len(df_day.index) > 2: # need at least 3 dates
return pd.infer_freq(df_day.index)
return 1
class TouEnergyChargeTariff(TimeOfUseTariff):
"""
This class represents a Time Of Use Energy Charge tariff
"""
def __init__(self, dates, time_schedule, unit_metric=TariffElemMetricUnit.EN_KWH, unit_cost=TariffElemCostUnit.DOLLAR, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param time_schedule: TODO
:param name: see FixedTariff init
"""
super(TouEnergyChargeTariff, self).__init__(dates, time_schedule, unit_metric, unit_cost, name)
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the bill due to a TOU tariff
:param df: a pandas dataframe
:return: a tuple (float, float) -> (cost, tot_energy)
"""
# Iterates over the days
energy = 0.0
cost = 0.0
# TODO: check for blockrate instead of assuming it's a float !
for idx, df_day in df.groupby(df.index.date):
daily_rate = self.rate_schedule.get_daily_rate(df_day.index[0])
df_prices = self.get_daily_price_dataframe(daily_rate, df_day)
# Unit and cost scale
mult_energy_unit = float(self.unit_metric.value)
mult_cost_unit = float(self.unit_cost.value)
# Cumulate the energy over the month
if data_col is not None:
df_values_in_day = df_day.loc[:, data_col]
else:
df_values_in_day = df_day[:]
energy += sum(df_values_in_day) / mult_energy_unit
# Cumulate the bill over the month
cost += sum(mult_cost_unit * df_values_in_day.multiply(df_prices.loc[:, 'price'].tolist())) / mult_energy_unit
return energy, cost
PK ! V: : J electricitycostcalculator_gabetest/cost_calculator/tariff_structure.py.bak__author__ = 'Olivier Van Cutsem'
from abc import abstractmethod
from enum import Enum
from datetime import datetime
import calendar
import pandas as pd
# --------------- TARIFF structures --------------- #
class TariffType(Enum):
FIX_CUSTOM_CHARGE = 'customer_fix_charge'
ENERGY_CUSTOM_CHARGE = 'customer_energy_charge'
DEMAND_CUSTOM_CHARGE_SEASON = 'customer_demand_charge_season'
DEMAND_CUSTOM_CHARGE_TOU = 'customer_demand_charge_tou'
PDP_ENERGY_CHARGE = 'pdp_event_energy_charge'
PDP_ENERGY_CREDIT = 'pdp_non_event_energy_credit'
PDP_DEMAND_CREDIT = 'pdp_non_event_demand_credit'
class TariffElemPeriod(Enum):
MONTHLY = 'M'
DAILY = 'D'
HOURLY = '1h'
HALFLY = '30min'
QUARTERLY = '15min'
class TariffElemMetricUnit(Enum):
EN_WH = 1
DEMAND_W = 1
EN_KWH = 1000.0
DEMAND_KW = 1000.0
class TariffElemCostUnit(Enum):
CENT = 0.01
DOLLAR = 1
class TariffBase(object):
"""
This abstract class represent the base of any tariffication structure.
The main components are the starting and ending date of the structure.
"""
def __init__(self, dates, unit_cost, name=None):
# Starting and ending dates, as timestamps
ts, te = dates
self.__startdate = ts
self.__enddate = te
self.name = name
self.unit_cost = unit_cost
def compute_bill(self, df, data_col=None):
"""
Compute the bill due to the power/energy consumption in df, for each billing period specified in billing_periods
It outputs a dictionary formatted as follow:
{
"bill_period_label1": (float or dict, float), -> the monthly 'metric' and its cost
...
}
:param df: a pandas dataframe containing power consumption timeseries
:param billing_periods: a dictionary mapping the billing periods label to a tuple (t_start, t_end) of datetime,
defining the period related to the billing label
:return: a dictionary formatted as in this method signature
"""
ret = {}
# Select only the data in this tariff window
start_sel = self.startdate
start_sel = start_sel.replace(tzinfo=df.index[0].tzinfo)
end_sel = self.enddate
end_sel = end_sel.replace(tzinfo=df.index[0].tzinfo)
mask = (df.index >= start_sel) & (df.index <= end_sel)
df = df.loc[mask]
# Loop over the months
t_s = df.index[0]
last_day_of_month = calendar.monthrange(t_s.year, t_s.month)[1] # The last day of this month
t_e = datetime(t_s.year, t_s.month, last_day_of_month, hour=23, minute=59, second=59, tzinfo=t_s.tzinfo) # end of the current month
t_e = min(df.index[-1], t_e)
while t_s <= t_e:
mask = (df.index >= t_s) & (df.index <= t_e)
df_month = df.loc[mask]
monthly_bill = self.compute_monthly_bill(df_month, data_col)
ret[t_s.strftime("%Y-%m")] = monthly_bill
# Prepare the next billing month
month = t_e.month + 1
year = t_e.year
if month >= 13:
month = 1
year += 1
t_s = datetime(year, month, 1, hour=0, minute=0, second=0, tzinfo=t_s.tzinfo)
last_day_of_month = calendar.monthrange(year, month)[1]
t_e = datetime(year, month, last_day_of_month, hour=23, minute=59, second=59, tzinfo=t_s.tzinfo)
t_e = min(df.index[-1], t_e)
return ret
@abstractmethod
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the monthly bill due to the power/energy consumption in df
:param df: a pandas dataframe
:param data_col: the column label containing the data
:return: a tuple (float, float) -> (value, cost), representing the bill and the corresponding metric linked to the cost
"""
pass
@property
def startdate(self):
"""
GETTER of the tariff starting date
:return: a timestamp
"""
return self.__startdate
@property
def enddate(self):
"""
GETTER of the tariff end date
:return: a timestamp
"""
return self.__enddate
@abstractmethod
def period_metric(self):
pass
@abstractmethod
def get_price_from_timestamp(self, timestamp):
pass
# --------------- FIXED TARIFF --------------- #
class FixedTariff(TariffBase):
"""
Represent a tariff fixed over a given period (among TariffPeriod)
"""
def __init__(self, dates, rate_value, unit_cost=TariffElemCostUnit.DOLLAR, bill_period=TariffElemPeriod.MONTHLY, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param bill_period: the period
:param name: see FixedTariff init
"""
super(FixedTariff, self).__init__(dates, unit_cost, name)
self.__rate_period = bill_period
self.__rate_value = rate_value
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the monthly bill due to a fixed periodic cost
:param df: a pandas dataframe
:return: a tuple (float, float), representing the bill and the duration (in months)
"""
first_day = df.index[0].day
last_day = df.index[-1].day
nb_days = last_day - first_day + 1
nb_days_per_month = 365/12
bill = 0
if self.__rate_period == TariffElemPeriod.MONTHLY:
bill = self.__rate_value * nb_days/nb_days_per_month # a fraction of the month
elif self.__rate_period == TariffElemPeriod.DAILY:
bill = self.__rate_value * nb_days # sum of each day
return nb_days, bill
def period_metric(self):
return self.__rate_period
def get_price_from_timestamp(self, timestamp):
return self.__rate_value
# --------------- TOU TARIFFs --------------- #
class TimeOfUseTariff(TariffBase):
"""
This class represents a tariff fixed over a given period (among TariffElemPeriod)
"""
def __init__(self, dates, rate_schedule, unit_metric, unit_cost, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param rate_list:
:param time_schedule: TODO
:param name: see FixedTariff init
"""
super(TimeOfUseTariff, self).__init__(dates, unit_cost, name)
self.__schedule = rate_schedule # A table mapping (month, day) to hourly rate index
self.__unit_metric = unit_metric
@abstractmethod
def compute_monthly_bill(self, df, data_col=None):
"""
idem super
"""
pass
@property
def rate_schedule(self):
return self.__schedule
@property
def unit_metric(self):
return self.__unit_metric
def period_metric(self):
nb_periods_in_day = self.__schedule.periods_in_day
# TODO: replace ifs by map
if nb_periods_in_day == 24:
return TariffElemPeriod.HOURLY
elif nb_periods_in_day == 24 * 2:
return TariffElemPeriod.HALFLY
elif nb_periods_in_day == 24 * 4:
return TariffElemPeriod.QUARTERLY
else:
return TariffElemPeriod.DAILY
def get_price_from_timestamp(self, timestamp):
return self.__schedule.get_from_timestamp(timestamp)
@staticmethod
def get_daily_price_dataframe(daily_rate, df_day):
# Constructing the dataframe for an easier manipulation of time
period_rate = len(daily_rate) / 24.0
# In some cases the day might not be full: missing data or DST
daily_prices = [daily_rate[int((df_day.index[i].hour + df_day.index[i].minute / 60.0) * period_rate)] for i in range(len(df_day.index))]
data = {'date': df_day.index[:], 'price': daily_prices}
df_prices = pd.DataFrame(data=data)
df_prices.set_index('date')
return df_prices
class TouDemandChargeTariff(TimeOfUseTariff):
"""
This class represents a Time Of Use Demand Charge tariff
"""
def __init__(self, dates, time_schedule, unit_metric=TariffElemMetricUnit.DEMAND_KW, unit_cost=TariffElemCostUnit.DOLLAR, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param rate_list: TODO
:param time_schedule: TODO
:param name: see FixedTariff init
"""
super(TouDemandChargeTariff, self).__init__(dates, time_schedule, unit_metric, unit_cost, name)
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the bill due to a TOU tariff
:param df: a pandas dataframe
:return: a tuple (dict, float) -> ({p1: (max_power_p1, time_max_p1), p2: (max_power_p2, time_max_p2), cost)
"""
# Scaling the power unit and cost
metric_unit_mult = float(self.unit_metric.value)
metric_price_mult = float(self.unit_cost.value)
max_per_set = {}
# df is in kWh and demand in kW: convert to Power
timestep_data = self.get_pd_timestep_data(df)
for idx, df_day in df.groupby(df.index.date):
daily_rate = self.rate_schedule.get_daily_rate(df_day.index[0])
set_of_daily_prices = set(daily_rate)
df_prices = self.get_daily_price_dataframe(daily_rate, df_day)
power_coeff = 1
if timestep_data == '15T':
power_coeff = 4
elif timestep_data == '30T':
power_coeff = 2
elif timestep_data == '60T' or timestep_data == 'H':
power_coeff = 1
for day_p in set_of_daily_prices:
# Create the mask in the day for this price
mask_price = df_prices['price'] == day_p
mask_price = mask_price.tolist()
date_max_period = None
if data_col is not None:
df_masked = df_day.loc[mask_price, data_col]
if len(df_masked) > 0:
date_max_period = df_masked.idxmax()
else:
df_masked = df_day.loc[mask_price]
if len(df_masked) > 0:
date_max_period = df_masked.idxmax()
if date_max_period is None:
continue
if data_col is not None:
max_power_period = df_day.loc[date_max_period, data_col] / metric_unit_mult
else:
max_power_period = df_day[date_max_period] / metric_unit_mult
max_power_period *= power_coeff # from kWh to kW
# Search for the same mask and update the value if a new mask
mask_price24h = mask_price
if len(mask_price) != len(daily_rate): # if this price is already in the list, take the corresponding mask ...
price_key = metric_price_mult * day_p
if price_key in max_per_set.keys():
mask_price24h = max_per_set[price_key]['mask']
add_this_demand = True
existing_mask_price = [k for k, v in max_per_set.items() if v['mask'] == mask_price24h]
if len(existing_mask_price) > 0: # Find the identical mask over the day
existing_mask_price = existing_mask_price[0]
if max_power_period > max_per_set[existing_mask_price]['max-demand']: # Check if the corresponding demand is greater
del max_per_set[existing_mask_price] # delete the former value, and add it (after)
else:
add_this_demand = False
# This is the first time this mask is seen OR this new demand is higher than the corresponding former: add it
if add_this_demand:
if type(date_max_period) is not pd.Timestamp:
max_power_date = date_max_period[data_col].to_pydatetime()
else:
max_power_date = date_max_period.to_pydatetime()
# The mask must be 24 hour long:
# it might not be the case for the first and last day, and the DST
price_key = metric_price_mult * day_p
max_per_set[price_key] = {'mask': mask_price24h, 'max-demand': max_power_period, 'max-demand-date': max_power_date}
return max_per_set
def get_pd_timestep_data(self, df):
"""
Return the most likely data frequency
:return:
"""
freq = 1
# Dataframe is complete
if df.index.freq is not None:
return df.index.freq
# Days are missing: loop through the day until getting a frequency
for idx, df_day in df.groupby(df.index.date):
if df_day.index.freq is not None:
return df_day.index.freq
elif len(df_day.index) > 2: # need at least 3 dates
return pd.infer_freq(df_day.index)
return 1
class TouEnergyChargeTariff(TimeOfUseTariff):
"""
This class represents a Time Of Use Energy Charge tariff
"""
def __init__(self, dates, time_schedule, unit_metric=TariffElemMetricUnit.EN_KWH, unit_cost=TariffElemCostUnit.DOLLAR, name=None):
"""
Constructor
:param dates: see FixedTariff init
:param time_schedule: TODO
:param name: see FixedTariff init
"""
super(TouEnergyChargeTariff, self).__init__(dates, time_schedule, unit_metric, unit_cost, name)
def compute_monthly_bill(self, df, data_col=None):
"""
Compute the bill due to a TOU tariff
:param df: a pandas dataframe
:return: a tuple (float, float) -> (cost, tot_energy)
"""
# Iterates over the days
energy = 0.0
cost = 0.0
# TODO: check for blockrate instead of assuming it's a float !
for idx, df_day in df.groupby(df.index.date):
daily_rate = self.rate_schedule.get_daily_rate(df_day.index[0])
df_prices = self.get_daily_price_dataframe(daily_rate, df_day)
# Unit and cost scale
mult_energy_unit = float(self.unit_metric.value)
mult_cost_unit = float(self.unit_cost.value)
# Cumulate the energy over the month
if data_col is not None:
df_values_in_day = df_day.loc[:, data_col]
else:
df_values_in_day = df_day[:]
energy += sum(df_values_in_day) / mult_energy_unit
# Cumulate the bill over the month
cost += sum(mult_cost_unit * df_values_in_day.multiply(df_prices.loc[:, 'price'].tolist())) / mult_energy_unit
return energy, cost
PK ! #3 3 9 electricitycostcalculator_gabetest/oadr_signal/.gitignore*.pyc
events.csv
*.xml
signals/*.xml
settings.json
PK ! s
= electricitycostcalculator_gabetest/oadr_signal/DR_template.pydef getSignalString(requestId, vtnId, eventId, modificationNumber, createdDateTime, eventStatus, vtnComment,
startTime, duration, signals, group, groupId=None, resourceId=None):
xml = '\n'
xml = xml + '\n'
xml = xml + '\t\n'
xml = xml + '\t\t\n'
xml = xml + '\t\t\t%s\n' % requestId
xml = xml + '\t\t\t%s\n' % vtnId
xml = xml + '\t\t\t\n'
xml = xml + '\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t%s\n' % eventId
xml = xml + '\t\t\t\t\t\t%d\n' % modificationNumber
xml = xml + '\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t%s\n' % createdDateTime
xml = xml + '\t\t\t\t\t\t%s\n' % eventStatus
xml = xml + '\t\t\t\t\t\t%s\n' % vtnComment
xml = xml + '\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t%s\n' % startTime
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t%s\n' % duration
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\n'
for signal in signals:
tagName = 'currencyPerKWh'
# default signalType = 'energy'
if signal['signalType'] == 'demand':
tagName = 'currencyPerKW'
itemUnits = signal['itemUnits']
scaleCode = signal['scaleCode']
eventHours = signal['eventHours']
prices = signal['prices']
signalName = signal['signalName']
signalId = signal['signalId']
currentPrice = signal['currentPrice']
xml = xml + '\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\n' % tagName
xml = xml + '\t\t\t\t\t\t\t\t%s\n' % tagName
xml = xml + '\t\t\t\t\t\t\t\t%s\n' % itemUnits
xml = xml + '\t\t\t\t\t\t\t\t%s\n' % scaleCode
xml = xml + '\t\t\t\t\t\t\t\n' % tagName
xml = xml + '\t\t\t\t\t\t\t\n'
for i in range(len(eventHours)):
xml = xml + '\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\t%s\n' % (eventHours[i])
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\tPT60M\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\t%d\n' % (i)
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\t\t%.2f\n' % (float(prices[i]))
xml = xml + '\t\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t%s\n' % signalName
xml = xml + '\t\t\t\t\t\t\tprice\n'
xml = xml + '\t\t\t\t\t\t\t%s\n' % signalId
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\t\t%.2f\n' % currentPrice
xml = xml + '\t\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\t\n'
if group:
xml = xml + '\t\t\t\t\t\t%s\n' % groupId
else:
xml = xml + '\t\t\t\t\t\t%s\n' % resourceId
xml = xml + '\t\t\t\t\t\n'
xml = xml + '\t\t\t\t\n'
xml = xml + '\t\t\t\n'
xml = xml + '\t\t\n'
xml = xml + '\t\n'
xml = xml + '\n'
return (xml)PK ! ! A electricitycostcalculator_gabetest/oadr_signal/DR_template.py.bakdef getSignalString(requestId, vtnId, eventId, modificationNumber, createdDateTime, eventStatus, vtnComment,
startTime, duration, signals, group, groupId=None, resourceId=None):
xml = u'\n'
xml = xml + u'\n'
xml = xml + u'\t\n'
xml = xml + u'\t\t\n'
xml = xml + u'\t\t\t%s\n' % requestId
xml = xml + u'\t\t\t%s\n' % vtnId
xml = xml + u'\t\t\t\n'
xml = xml + u'\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t%s\n' % eventId
xml = xml + u'\t\t\t\t\t\t%d\n' % modificationNumber
xml = xml + u'\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t%s\n' % createdDateTime
xml = xml + u'\t\t\t\t\t\t%s\n' % eventStatus
xml = xml + u'\t\t\t\t\t\t%s\n' % vtnComment
xml = xml + u'\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t%s\n' % startTime
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t%s\n' % duration
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\n'
for signal in signals:
tagName = 'currencyPerKWh'
# default signalType = 'energy'
if signal['signalType'] == 'demand':
tagName = 'currencyPerKW'
itemUnits = signal['itemUnits']
scaleCode = signal['scaleCode']
eventHours = signal['eventHours']
prices = signal['prices']
signalName = signal['signalName']
signalId = signal['signalId']
currentPrice = signal['currentPrice']
xml = xml + u'\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\n' % tagName
xml = xml + u'\t\t\t\t\t\t\t\t%s\n' % tagName
xml = xml + u'\t\t\t\t\t\t\t\t%s\n' % itemUnits
xml = xml + u'\t\t\t\t\t\t\t\t%s\n' % scaleCode
xml = xml + u'\t\t\t\t\t\t\t\n' % tagName
xml = xml + u'\t\t\t\t\t\t\t\n'
for i in range(len(eventHours)):
xml = xml + u'\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\t%s\n' % (eventHours[i])
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\tPT60M\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\t%d\n' % (i)
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\t\t%.2f\n' % (float(prices[i]))
xml = xml + u'\t\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t%s\n' % signalName
xml = xml + u'\t\t\t\t\t\t\tprice\n'
xml = xml + u'\t\t\t\t\t\t\t%s\n' % signalId
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\t\t%.2f\n' % currentPrice
xml = xml + u'\t\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\t\n'
if group:
xml = xml + u'\t\t\t\t\t\t%s\n' % groupId
else:
xml = xml + u'\t\t\t\t\t\t%s\n' % resourceId
xml = xml + u'\t\t\t\t\t\n'
xml = xml + u'\t\t\t\t\n'
xml = xml + u'\t\t\t\n'
xml = xml + u'\t\t\n'
xml = xml + u'\t\n'
xml = xml + u'\n'
return (xml)PK ! -b&