import logging
from collections.abc import Mapping
from datetime import datetime
from typing import Dict, Iterable, List, Optional, Tuple
import pandas as pd
LOGGER = logging.getLogger(__name__)
[docs]class ResultsPackage(Mapping):
"""A package of OSeMOSYS results
Internal data structure is a dictionary of pandas DataFrames
A crude caching function is implemented whereby calculated results are stored in the
internal data structure so that each result is only calculated once.
Arguments
---------
data : dict
A dictionary of results data
input_data: dict, default=None
Dictionary of input data
"""
def __init__(
self,
data: Dict[str, pd.DataFrame],
input_data: Optional[Dict[str, pd.DataFrame]] = None,
):
super().__init__()
self._data = data
if input_data:
self._package = input_data
else:
self._package = {}
self._result_mapper = {
"AccumulatedNewCapacity": self.accumulated_new_capacity,
"AnnualEmissions": self.annual_emissions,
"AnnualFixedOperatingCost": self.annual_fixed_operating_cost,
"AnnualTechnologyEmission": self.annual_technology_emissions,
"AnnualTechnologyEmissionByMode": self.annual_technology_emission_by_mode,
"AnnualVariableOperatingCost": self.annual_variable_operating_cost,
"CapitalInvestment": self.capital_investment,
"Demand": self.demand,
"DiscountedTechnologyEmissionsPenalty": self.discounted_tech_emis_pen,
"ProductionByTechnology": self.production_by_technology,
"ProductionByTechnologyAnnual": self.production_by_technology_annual,
"RateOfProductionByTechnology": self.rate_of_product_technology,
"RateOfProductionByTechnologyByMode": self.rate_of_production_tech_mode,
"RateOfUseByTechnology": self.rate_of_use_by_technology,
"RateOfUseByTechnologyByMode": self.rate_of_use_by_technology_by_mode,
"TotalAnnualTechnologyActivityByMode": self.total_annual_tech_activity_mode,
"TotalCapacityAnnual": self.total_capacity_annual,
"TotalDiscountedCost": self.total_discounted_cost,
"TotalTechnologyAnnualActivity": self.total_technology_annual_activity,
"TotalTechnologyModelPeriodActivity": self.total_tech_model_period_activity,
"UseByTechnology": self.use_by_technology,
}
self._result_cache = {} # type: Dict[str, pd.DataFrame]
@property
def data(self) -> Dict[str, pd.DataFrame]:
"""View the results dictionary"""
return self._data
@property
def result_mapper(self) -> Dict[str, pd.DataFrame]:
return self._result_mapper
@property
def result_cache(self) -> Dict[str, pd.DataFrame]:
return self._result_cache
@result_cache.setter
def result_cache(self, value: Iterable[Tuple[str, pd.DataFrame]]):
self._result_cache.update(value)
def __getitem__(self, name: str) -> pd.DataFrame:
LOGGER.debug("Returning '%s' from ... ", name)
if name in self.data.keys():
LOGGER.debug(" ... ResultsPackage.data")
return self.data[name]
elif name in self._package.keys():
LOGGER.debug(" ... ResultsPackage.input_data")
return self._package[name]
elif name in self.result_cache.keys():
LOGGER.debug(" ... ResultsPackage.result_cache")
return self.result_cache[name]
elif name in self.result_mapper.keys():
# Implements a crude form of caching, where calculated results are
# first stored in the internal dict, and then returned
LOGGER.debug(" ... ResultsPackage.calculating ...")
start = datetime.now()
results = self.result_mapper[name]()
stop = datetime.now()
diff = stop - start
total = diff.total_seconds()
LOGGER.debug("Calculation took %s secs", total)
LOGGER.debug("Caching results for %s", name)
self.result_cache[name] = results
return self.result_cache[name]
else:
LOGGER.debug(" ... Not found in cache or calculation methods")
raise KeyError("{} is not accessible or available".format(name))
return self.data[name]
def __iter__(self):
raise NotImplementedError()
def __len__(self):
raise NotImplementedError()
def _msg(self, name: str, error: str):
return "Cannot calculate {} due to missing data: {}".format(name, error)
[docs] def accumulated_new_capacity(self) -> pd.DataFrame:
"""AccumulatedNewCapacity
Arguments
---------
operational_life: pandas.DataFrame
new_capacity: pandas.DataFrame
year: pandas.Index
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, y~YEAR,
sum{yy in YEAR: y-yy < OperationalLife[r,t] && y-yy>=0}
NewCapacity[r,t,yy] ~VALUE;
"""
try:
new_capacity = self["NewCapacity"].copy()
year = pd.Index(self["YEAR"]["VALUE"].to_list())
except KeyError as ex:
raise KeyError(self._msg("AccumulatedNewCapacity", str(ex)))
new_capacity["OperationalLife"] = self["OperationalLife"].copy()
regions = new_capacity.reset_index()["REGION"].unique()
technologies = new_capacity.reset_index()["TECHNOLOGY"].unique()
index = pd.MultiIndex.from_product(
[regions, technologies, year.to_list()],
names=["REGION", "TECHNOLOGY", "YEAR"],
)
acc_capacity = new_capacity.reindex(index, copy=True)
for index, data in new_capacity.reset_index().groupby(
by=["REGION", "TECHNOLOGY"]
):
region, technology = index
for yr in year:
mask = (yr - data["YEAR"] < data["OperationalLife"]) & (
yr - data["YEAR"] >= 0
)
acc_capacity.loc[region, technology, yr] = data[mask].sum()
acc_capacity = acc_capacity.drop(columns="OperationalLife")
return acc_capacity[(acc_capacity != 0).all(1)]
[docs] def annual_emissions(self) -> pd.DataFrame:
"""Calculates the annual emissions
Annual Emission are found by multiplying the emission activity ratio
(emissions per unit activity) by the rate of activity and the yearsplit
Notes
-----
From the formulation::
sum{t in TECHNOLOGY, l in TIMESLICE, m in MODE_OF_OPERATION:
EmissionActivityRatio[r,t,e,m,y]<>0}
RateOfActivity[r,l,t,m,y] * EmissionActivityRatio[r,t,e,m,y]
* YearSplit[l,y]~VALUE;
"""
try:
emission_activity_ratio = self["EmissionActivityRatio"]
yearsplit = self["YearSplit"]
rate_of_activity = self["RateOfActivity"]
except KeyError as ex:
raise KeyError(self._msg("AnnualEmissions", str(ex)))
mid = emission_activity_ratio.mul(yearsplit)
data = mid.mul(rate_of_activity, fill_value=0.0)
if not data.empty:
data = data.groupby(by=["REGION", "EMISSION", "YEAR"]).sum()
return data
[docs] def annual_fixed_operating_cost(self) -> pd.DataFrame:
"""Compute AnnualFixedOperatingCost result
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, y~YEAR,
FixedCost[r,t,y] *
((sum{yy in YEAR: y-yy < OperationalLife[r,t] && y-yy>=0}
NewCapacity[r,t,yy]) + ResidualCapacity[r,t,y]) ~VALUE;
"""
try:
total_capacity = self["TotalCapacityAnnual"]
fixed_cost = self["FixedCost"]
except KeyError as ex:
raise KeyError(self._msg("AnnualFixedOperatingCost", str(ex)))
total_fixed_costs = total_capacity.mul(fixed_cost, fill_value=0.0)
return total_fixed_costs[(total_fixed_costs != 0).all(1)].dropna()
[docs] def annual_technology_emissions(self) -> pd.DataFrame:
"""Calculates results ``AnnualTechnologyEmission``
Notes
-----
From the formulation::
REGION, TECHNOLOGY, EMISSION, YEAR,
sum{l in TIMESLICE, m in MODE_OF_OPERATION:
EmissionActivityRatio[r,t,e,m,y]<>0}
EmissionActivityRatio[r,t,e,m,y] * RateOfActivity[r,l,t,m,y]
* YearSplit[l,y];
"""
try:
data = self["AnnualTechnologyEmissionByMode"].copy(deep=True)
except KeyError as ex:
raise KeyError(self._msg("AnnualTechnologyEmission", str(ex)))
if not data.empty:
data = data.groupby(by=["REGION", "TECHNOLOGY", "EMISSION", "YEAR"]).sum()
return data[(data != 0).all(1)]
[docs] def annual_technology_emission_by_mode(self) -> pd.DataFrame:
"""AnnualTechnologyEmissionByMode
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, e~EMISSION, m~MODE_OF_OPERATION, y~YEAR,
sum{l in TIMESLICE: EmissionActivityRatio[r,t,e,m,y] <> 0}
EmissionActivityRatio[r,t,e,m,y] * RateOfActivity[r,l,t,m,y]
* YearSplit[l,y]
"""
try:
emission_activity_ratio: pd.DataFrame = self["EmissionActivityRatio"]
yearsplit: pd.DataFrame = self["YearSplit"]
rate_of_activity: pd.DataFrame = self["RateOfActivity"]
except KeyError as ex:
raise KeyError(self._msg("AnnualTechnologyEmissionByMode", str(ex)))
mid = emission_activity_ratio.mul(yearsplit)
data = mid.mul(rate_of_activity)
if not data.empty:
data = data.groupby(
by=["REGION", "TECHNOLOGY", "EMISSION", "MODE_OF_OPERATION", "YEAR"]
).sum()
return data[(data != 0).all(1)]
[docs] def annual_variable_operating_cost(self) -> pd.DataFrame:
"""AnnualVariableOperatingCost
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, y~YEAR,
sum{m in MODE_OF_OPERATION, l in TIMESLICE}
RateOfActivity[r,l,t,m,y]
* YearSplit[l,y]
* VariableCost[r,t,m,y] ~VALUE;
"""
try:
rate_of_activity = self["RateOfActivity"]
yearsplit = self["YearSplit"]
variable_cost = self["VariableCost"]
except KeyError as ex:
raise KeyError(self._msg("AnnualVariableOperatingCost", str(ex)))
split_activity = rate_of_activity.mul(yearsplit, fill_value=0.0)
data = split_activity.mul(variable_cost, fill_value=0.0)
if not data.empty:
data = data.groupby(by=["REGION", "TECHNOLOGY", "YEAR"]).sum()
return data[(data != 0).all(1)]
[docs] def capital_investment(self) -> pd.DataFrame:
"""CapitalInvestment
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, y~YEAR,
CapitalCost[r,t,y] * NewCapacity[r,t,y] * CapitalRecoveryFactor[r,t] *
PvAnnuity[r,t] ~VALUE;
"""
try:
capital_cost = self["CapitalCost"]
new_capacity = self["NewCapacity"]
operational_life = self["OperationalLife"]
discount_rate = self["DiscountRate"]
discount_rate_idv = self["DiscountRateIdv"]
regions = self["REGION"]["VALUE"].to_list()
technologies = self.get_unique_values_from_index(
[
capital_cost,
new_capacity,
],
"TECHNOLOGY",
)
except KeyError as ex:
raise KeyError(self._msg("CapitalInvestment", str(ex)))
crf = capital_recovery_factor(
regions, technologies, discount_rate_idv, operational_life
)
pva = pv_annuity(regions, technologies, discount_rate, operational_life)
capital_investment = capital_cost.mul(new_capacity, fill_value=0.0)
capital_investment = capital_investment.mul(crf, fill_value=0.0).mul(
pva, fill_value=0.0
)
data = capital_investment
if not data.empty:
data = data.groupby(by=["REGION", "TECHNOLOGY", "YEAR"]).sum()
return data[(data != 0).all(1)]
[docs] def demand(self) -> pd.DataFrame:
"""Demand
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, f~FUEL, y~YEAR,
SpecifiedAnnualDemand[r,f,y] * SpecifiedDemandProfile[r,f,l,y] ~VALUE;
"""
try:
specified_annual_demand = self["SpecifiedAnnualDemand"]
specified_demand_profile = self["SpecifiedDemandProfile"]
except KeyError as ex:
raise KeyError(self._msg("Demand", str(ex)))
data = specified_annual_demand.mul(specified_demand_profile, fill_value=0.0)
if not data.empty:
data = data.reset_index().set_index(["REGION", "TIMESLICE", "FUEL", "YEAR"])
return data[(data != 0).all(1)]
[docs] def discounted_tech_emis_pen(self) -> pd.DataFrame:
"""DiscountedTechnologyEmissionsPenalty
Notes
-----
From the formulation::
DiscountedTechnologyEmissionsPenalty[r,t,y] :=
EmissionActivityRatio[r,t,e,m,y] * RateOfActivity[r,l,t,m,y] *
YearSplit[l,y] * EmissionsPenalty[r,e,y] / DiscountFactorMid[r,y]
"""
try:
annual_technology_emission_by_mode = self["AnnualTechnologyEmissionByMode"]
emission_penalty = self["EmissionsPenalty"]
regions = self["REGION"]["VALUE"].to_list()
years = self["YEAR"]["VALUE"].to_list()
discount_rate = self["DiscountRate"]
except KeyError as ex:
raise KeyError(self._msg("DiscountedTechnologyEmissionsPenalty", str(ex)))
discount_factor_mid = discount_factor(regions, years, discount_rate, 0.5)
emissions_penalty = annual_technology_emission_by_mode.mul(
emission_penalty, fill_value=0.0
)
data = emissions_penalty.div(discount_factor_mid, fill_value=0.0)
if not data.empty:
data = data.groupby(by=["REGION", "TECHNOLOGY", "YEAR"]).sum()
return data[(data != 0).all(1)]
[docs] def production_by_technology(self) -> pd.DataFrame:
"""ProductionByTechnology
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, t~TECHNOLOGY, f~FUEL, y~YEAR,
sum{m in MODE_OF_OPERATION: OutputActivityRatio[r,t,f,m,y] <> 0}
RateOfActivity[r,l,t,m,y] * OutputActivityRatio[r,t,f,m,y]
* YearSplit[l,y] ~VALUE;
"""
try:
rate_of_activity = self["RateOfActivity"]
output_activity_ratio = self["OutputActivityRatio"]
year_split = self["YearSplit"]
except KeyError as ex:
raise KeyError(self._msg("ProductionByTechnology", str(ex)))
split_activity = rate_of_activity.mul(year_split, fill_value=0.0)
data = split_activity.mul(output_activity_ratio, fill_value=0.0)
if not data.empty:
data = data.groupby(
by=["REGION", "TIMESLICE", "TECHNOLOGY", "FUEL", "YEAR"]
).sum()
return data[(data != 0).all(1)]
[docs] def production_by_technology_annual(self) -> pd.DataFrame:
"""Aggregates production by technology to the annual level"""
try:
production_by_technology = self["ProductionByTechnology"].copy(deep=True)
except KeyError as ex:
raise KeyError(self._msg("ProductionByTechnologyAnnual", str(ex)))
data = production_by_technology
if not data.empty:
data = data.groupby(by=["REGION", "TECHNOLOGY", "FUEL", "YEAR"]).sum()
return data[(data != 0).all(1)]
[docs] def rate_of_production_tech_mode(self) -> pd.DataFrame:
"""RateOfProductionByTechnologyByMode
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, t~TECHNOLOGY, m~MODE_OF_OPERATION, f~FUEL, y~YEAR,
RateOfActivity[r,l,t,m,y] * OutputActivityRatio[r,t,f,m,y]~VALUE;
"""
try:
rate_of_activity = self["RateOfActivity"]
output_activity_ratio = self["OutputActivityRatio"]
except KeyError as ex:
raise KeyError(self._msg("RateOfProductionByTechnologyByMode", str(ex)))
data = rate_of_activity.mul(output_activity_ratio, fill_value=0.0)
if not data.empty:
data = data.reset_index().set_index(
[
"REGION",
"TIMESLICE",
"TECHNOLOGY",
"MODE_OF_OPERATION",
"FUEL",
"YEAR",
]
)
return data[(data != 0).all(1)].sort_index()
[docs] def rate_of_product_technology(self) -> pd.DataFrame:
"""Sums up mode of operation for rate of production
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, t~TECHNOLOGY, f~FUEL, y~YEAR,
sum{m in MODE_OF_OPERATION: OutputActivityRatio[r,t,f,m,y] <> 0}
RateOfActivity[r,l,t,m,y] * OutputActivityRatio[r,t,f,m,y]~VALUE;
"""
try:
rate_of_production = self["RateOfProductionByTechnologyByMode"].copy(
deep=True
)
except KeyError as ex:
raise KeyError(self._msg("RateOfProductionByTechnology", str(ex)))
data = rate_of_production
if not data.empty:
data = data.groupby(
by=["REGION", "TIMESLICE", "TECHNOLOGY", "FUEL", "YEAR"]
).sum()
return data[(data != 0).all(1)].sort_index()
[docs] def rate_of_use_by_technology(self) -> pd.DataFrame:
"""RateOfUseByTechnology
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, t~TECHNOLOGY, f~FUEL, y~YEAR,
sum{m in MODE_OF_OPERATION: InputActivityRatio[r,t,f,m,y]<>0}
RateOfActivity[r,l,t,m,y] * InputActivityRatio[r,t,f,m,y]~VALUE;
"""
try:
rate_of_use_by_technology_by_mode = self[
"RateOfUseByTechnologyByMode"
].copy(deep=True)
except KeyError as ex:
raise KeyError(self._msg("RateOfUseByTechnology", str(ex)))
data = rate_of_use_by_technology_by_mode
if not data.empty:
data = data.groupby(
by=["REGION", "TIMESLICE", "TECHNOLOGY", "FUEL", "YEAR"]
).sum()
return data[(data != 0).all(1)]
[docs] def rate_of_use_by_technology_by_mode(self) -> pd.DataFrame:
"""RateOfUseByTechnologyByMode
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, t~TECHNOLOGY, m~MODE_OF_OPERATION, f~FUEL, y~YEAR,
RateOfActivity[r,l,t,m,y] * InputActivityRatio[r,t,f,m,y]~VALUE;
"""
try:
input_activity_ratio = self["InputActivityRatio"]
rate_of_activity = self["RateOfActivity"]
except KeyError as ex:
raise KeyError(self._msg("RateOfUseByTechnology", str(ex)))
data = input_activity_ratio.mul(rate_of_activity, fill_value=0.0)
return data[(data != 0).all(1)]
[docs] def total_annual_tech_activity_mode(self) -> pd.DataFrame:
"""TotalAnnualTechnologyActivityByMode
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, m~MODE_OF_OPERATION, y~YEAR,
sum{l in TIMESLICE}
RateOfActivity[r,l,t,m,y] * YearSplit[l,y]~VALUE;
"""
try:
rate_of_activity = self["RateOfActivity"]
year_split = self["YearSplit"]
except KeyError as ex:
raise KeyError(self._msg("TotalAnnualTechnologyActivityByMode", str(ex)))
data = rate_of_activity.mul(year_split, fill_value=0.0)
return data[(data != 0).all(1)]
[docs] def total_capacity_annual(self) -> pd.DataFrame:
"""TotalCapacityAnnual
Notes
-----
From the formulation::
r~REGION, t~TECHNOLOGY, y~YEAR,
ResidualCapacity[r,t,y] +
(sum{yy in YEAR: y-yy < OperationalLife[r,t] && y-yy>=0}
NewCapacity[r,t,yy])~VALUE;
"""
try:
residual_capacity = self["ResidualCapacity"]
acc_new_capacity = self["AccumulatedNewCapacity"]
except KeyError as ex:
raise KeyError(self._msg("TotalCapacityAnnual", str(ex)))
data = residual_capacity.add(acc_new_capacity, fill_value=0.0)
return data[(data != 0).all(1)]
[docs] def total_discounted_cost(self) -> pd.DataFrame:
"""TotalDiscountedCost
Notes
-----
From the formulation::
r~REGION, y~YEAR,
sum{t in TECHNOLOGY}
(
(
(
(
sum{yy in YEAR: y-yy < OperationalLife[r,t] && y-yy>=0}
NewCapacity[r,t,yy]
)
+ ResidualCapacity[r,t,y]
)
* FixedCost[r,t,y]
+ sum{l in TIMESLICE, m in MODEperTECHNOLOGY[t]}
RateOfActivity[r,l,t,m,y] * YearSplit[l,y] * VariableCost[r,t,m,y]
)
/ (DiscountFactorMid[r,y])
+ CapitalCost[r,t,y] * NewCapacity[r,t,y] * CapitalRecoveryFactor[r,t] * PvAnnuity[r,t] / (DiscountFactor[r,y])
+ DiscountedTechnologyEmissionsPenalty[r,t,y] - DiscountedSalvageValue[r,t,y])
+ sum{s in STORAGE}
(
CapitalCostStorage[r,s,y] * NewStorageCapacity[r,s,y] / (DiscountFactorStorage[r,s,y])
- CapitalCostStorage[r,s,y] * NewStorageCapacity[r,s,y] / (DiscountFactorStorage[r,s,y]
)
) ~VALUE;
"""
try:
discount_rate = self["DiscountRate"]
year_df = self["YEAR"].copy(deep=True)
region_df = self["REGION"].copy(deep=True)
years = year_df["VALUE"].tolist()
regions = region_df["VALUE"].tolist()
annual_fixed_operating_cost = self["AnnualFixedOperatingCost"]
annual_variable_operating_cost = self["AnnualVariableOperatingCost"]
capital_investment = self["CapitalInvestment"]
discounted_emissions_penalty = self["DiscountedTechnologyEmissionsPenalty"]
discounted_salvage_value = self["DiscountedSalvageValue"]
# capital_cost_storage = self["CapitalCostStorage"]
except KeyError as ex:
raise KeyError(self._msg("TotalDiscountedCost", str(ex)))
df_start = discount_factor(regions, years, discount_rate, 0.0)
df_mid = discount_factor(regions, years, discount_rate, 0.5)
undiscounted_operational_costs = annual_fixed_operating_cost.add(
annual_variable_operating_cost, fill_value=0.0
)
discounted_operational_costs = undiscounted_operational_costs.div(
df_mid, fill_value=0.0
)
discounted_capital_costs = capital_investment.div(df_start, fill_value=0.0)
discounted_total_costs = discounted_operational_costs.add(
discounted_capital_costs, fill_value=0.0
)
discounted_total_costs = discounted_total_costs.add(
discounted_emissions_penalty, fill_value=0.0
)
discounted_total_costs = discounted_total_costs.sub(
discounted_salvage_value, fill_value=0.0
)
data = discounted_total_costs
if not data.empty:
data = data.groupby(by=["REGION", "YEAR"]).sum()
return data[(data != 0).all(1)].dropna()
[docs] def get_unique_values_from_index(self, dataframes: List, name: str) -> List:
"""Utility function to extract list of unique values
Extract unique values from the same index of the passed dataframes
"""
elements = [] # type: List
for df in dataframes:
df = df.reset_index()
if name in df.columns:
elements += list(df[name].unique())
return list(set(elements))
[docs] def total_technology_annual_activity(self) -> pd.DataFrame:
"""TotalTechnologyAnnualActivity
Notes
-----
From the formulation::
ResultsPath & "/TotalTechnologyAnnualActivity.csv":
r~REGION, t~TECHNOLOGY, y~YEAR,
sum{l in TIMESLICE, m in MODE_OF_OPERATION}
RateOfActivity[r,l,t,m,y] * YearSplit[l,y]~VALUE;
"""
try:
data = self["TotalAnnualTechnologyActivityByMode"].copy(deep=True)
except KeyError as ex:
raise KeyError(self._msg("TotalTechnologyAnnualActivity", str(ex)))
if not data.empty:
data = data.groupby(["REGION", "TECHNOLOGY", "YEAR"]).sum()
return data[(data != 0).all(1)]
[docs] def total_tech_model_period_activity(self) -> pd.DataFrame:
"""TotalTechnologyModelPeriodActivity
Notes
-----
From the formulation::
ResultsPath & "/TotalTechnologyModelPeriodActivity.csv":
r~REGION, t~TECHNOLOGY,
sum{l in TIMESLICE, m in MODE_OF_OPERATION, y in YEAR}
RateOfActivity[r,l,t,m,y]*YearSplit[l,y]~VALUE;
"""
try:
data = self["TotalTechnologyAnnualActivity"].copy(deep=True)
except KeyError as ex:
raise KeyError(self._msg("TotalTechnologyModelPeriodActivity", str(ex)))
if not data.empty:
data = data.groupby(["REGION", "TECHNOLOGY"]).sum()
return data[(data != 0).all(1)]
[docs] def use_by_technology(self) -> pd.DataFrame:
"""UseByTechnology
Notes
-----
From the formulation::
r~REGION, l~TIMESLICE, t~TECHNOLOGY, f~FUEL, y~YEAR,
sum{m in MODE_OF_OPERATION}
RateOfActivity[r,l,t,m,y]
* InputActivityRatio[r,t,f,m,y]
* YearSplit[l,y]~VALUE;
"""
try:
rate_of_use = self["RateOfUseByTechnologyByMode"]
year_split = self["YearSplit"]
except KeyError as ex:
raise KeyError(self._msg("UseByTechnology", str(ex)))
data = rate_of_use.mul(year_split, fill_value=0.0)
if not data.empty:
data = data.groupby(
["REGION", "TIMESLICE", "TECHNOLOGY", "FUEL", "YEAR"]
).sum()
return data[(data != 0).all(1)]
[docs]def capital_recovery_factor(
regions: List,
technologies: List,
discount_rate_idv: pd.DataFrame,
operational_life: pd.DataFrame,
) -> pd.DataFrame:
"""Calculates the capital recovery factor
Arguments
---------
regions: list
technologies: list
discount_rate_idv: pd.DataFrame
operational_life: pd.DataFrame
Notes
-----
From the formulation::
param CapitalRecoveryFactor{r in REGION, t in TECHNOLOGY} :=
(1 - (1 + DiscountRateIdv[r,t])^(-1))/(1 - (1 + DiscountRateIdv[r,t])^(-(OperationalLife[r,t])));
"""
if regions and technologies:
index = pd.MultiIndex.from_product(
[regions, technologies], names=["REGION", "TECHNOLOGY"]
)
crf = discount_rate_idv.reindex(index)
crf["RATE"] = crf["VALUE"] + 1
crf["NUMER"] = 1 - crf["RATE"].pow(-1)
crf["DENOM"] = 1 - crf["RATE"].pow(-operational_life["VALUE"])
crf["VALUE"] = (crf["NUMER"] / crf["DENOM"]).round(6)
return crf.reset_index()[["REGION", "TECHNOLOGY", "VALUE"]].set_index(
["REGION", "TECHNOLOGY"]
)
else:
return pd.DataFrame([], columns=["REGION", "TECHNOLOGY", "VALUE"]).set_index(
["REGION", "TECHNOLOGY"]
)
[docs]def pv_annuity(
regions: List,
technologies: List,
discount_rate: pd.DataFrame,
operational_life: pd.DataFrame,
) -> pd.DataFrame:
"""Calculates the present value of an annuity
Arguments
---------
regions: list
technologies: list
discount_rate: pd.DataFrame
operational_life: pd.DataFrame
Notes
-----
From the formulation::
param PvAnnuity{r in REGION, t in TECHNOLOGY} :=
(1 - (1 + DiscountRate[r])^(-(OperationalLife[r,t]))) * (1 + DiscountRate[r]) / DiscountRate[r];
"""
if regions and technologies:
index = pd.MultiIndex.from_product(
[regions, technologies], names=["REGION", "TECHNOLOGY"]
)
pva = discount_rate.reindex(index).reset_index(level="TECHNOLOGY")
pva["RATE"] = discount_rate["VALUE"] + 1
pva = pva.set_index([pva.index, "TECHNOLOGY"])
pva["VALUE"] = (
(1 - pva["RATE"].pow(-operational_life["VALUE"])).mul(pva["RATE"])
/ discount_rate["VALUE"]
).round(6)
return pva.reset_index()[["REGION", "TECHNOLOGY", "VALUE"]].set_index(
["REGION", "TECHNOLOGY"]
)
else:
return pd.DataFrame([], columns=["REGION", "TECHNOLOGY", "VALUE"]).set_index(
["REGION", "TECHNOLOGY"]
)
[docs]def discount_factor(
regions: List,
years: List,
discount_rate: pd.DataFrame,
adj: float = 0.0,
) -> pd.DataFrame:
"""DiscountFactor
Arguments
---------
regions: list
years: list
discount_rate: pd.DataFrame
adj: float, default=0.0
Adjust to beginning of the year (default), mid year (0.5) or end year (1.0)
Notes
-----
From the formulation::
param DiscountFactor{r in REGION, y in YEAR} :=
(1 + DiscountRate[r]) ^ (y - min{yy in YEAR} min(yy) + 0.0);
param DiscountFactorMid{r in REGION, y in YEAR} :=
(1 + DiscountRate[r]) ^ (y - min{yy in YEAR} min(yy) + 0.5);
"""
if regions and years:
discount_rate["YEAR"] = [years]
discount_factor = discount_rate.explode("YEAR").reset_index(level="REGION")
discount_factor["YEAR"] = discount_factor["YEAR"].astype("int64")
discount_factor["NUM"] = discount_factor["YEAR"] - discount_factor["YEAR"].min()
discount_factor["RATE"] = discount_factor["VALUE"] + 1
discount_factor["VALUE"] = (
discount_factor["RATE"].pow(discount_factor["NUM"] + adj).astype(float)
)
return discount_factor.reset_index()[["REGION", "YEAR", "VALUE"]].set_index(
["REGION", "YEAR"]
)
else:
return pd.DataFrame([], columns=["REGION", "YEAR", "VALUE"]).set_index(
["REGION", "YEAR"]
)
[docs]def discount_factor_storage(
regions: List,
storages: List,
years: List,
discount_rate_storage: pd.DataFrame,
adj: float = 0.0,
) -> pd.DataFrame:
"""DiscountFactorStorage
Arguments
---------
regions: list
storages: list
years: list
discount_rate_storage: pd.DataFrame
adj: float, default=0.0
Adjust to beginning of the year (default), mid year (0.5) or end year (1.0)
Notes
-----
From the formulation::
param DiscountFactorStorage{r in REGION, s in STORAGE, y in YEAR} :=
(1 + DiscountRateStorage[r,s]) ^ (y - min{yy in YEAR} min(yy) + 0.0);
"""
if regions and years:
index = pd.MultiIndex.from_product(
[regions, storages, years], names=["REGION", "STORAGE", "YEAR"]
)
discount_fac_storage = discount_rate_storage.reindex(index).reset_index(
level="YEAR"
)
discount_fac_storage["NUM"] = (
discount_fac_storage["YEAR"] - discount_fac_storage["YEAR"].min()
)
discount_fac_storage["RATE"] = 1 + discount_rate_storage
discount_fac_storage["VALUE"] = discount_fac_storage["RATE"].pow(
discount_fac_storage["NUM"] + adj
)
return discount_fac_storage.reset_index()[
["REGION", "STORAGE", "YEAR", "VALUE"]
].set_index(["REGION", "STORAGE", "YEAR"])
else:
return pd.DataFrame(
[], columns=["REGION", "STORAGE", "YEAR", "VALUE"]
).set_index(["REGION", "STORAGE", "YEAR"])