diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..7dfa029 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,7 @@ +{ + "[python]": { + "editor.defaultFormatter": "ms-python.black-formatter", + "editor.formatOnSave": true + }, + "black-formatter.args": ["--line-length", "100"] +} diff --git a/README.md b/README.md index acbaae4..4b3e095 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,7 @@ Python files and Jupyter notebooks for processing the Annual Technology Baseline (ATB) electricity data and determining LCOE and other metrics. All documentation and data for the ATB is available at the [ATB website](https://atb.nrel.gov). ## Installation and Requirements + The pipeline requires [Python](https://www.python.org) 3.10 or newer. Dependancies can be installed using `pip`: ``` @@ -21,6 +22,7 @@ $ pytest Tests take about a minute and should complete without errors. The ATB pipeline uses [xlwings](https://www.xlwings.org/) for accessing the ATB data workbook and requires a copy of Microsoft Excel. Currently the full pipeline will only run on MacOS and Windows. Linux support may be added in the future. ## Running the ATB Electricity Pipeline + Running the pipeline requires downloaded the most current data in `xlsx` format from the [ATB website](https://atb.nrel.gov). The pipeline may be ran for one or all ATB electricity technologies. Data may be exported in several formats. Below are several example workflows. It is assumed that all @@ -54,6 +56,7 @@ $ python -m lcoe_calculator.process_all --help ``` ## Debt Fraction Calculator + The debt fraction calculator uses [PySAM](https://nrel-pysam.readthedocs.io/en/main/) to calculate debt fractions for one or all ATB technologies. To calculate debt fractions for all technologies run the following from the repository root directory: @@ -81,6 +84,7 @@ $ python -m debt_fraction_calculator.debt_fraction_calc --help ``` ## Example Jupyter Notebooks + The `./example_notebooks` directory has several [Jupyter](https://jupyter.org/) notebooks showing how to perform various tasks. The notebooks are a good way to understand how to use the code and experiment with the ATB pipeline code. Jupyter must first be installed before use: @@ -98,14 +102,20 @@ $ jupyter-notebook in the root repository directory. ## Notable Directories + - `./lcoe_calculator` Extract technology metrics from the data workbook `xlsx` file and calculate LCOE -using Python. + using Python. - `./debt_fraction_calculator` Given data and assumptions in the data workbook xlsx file, calculate -debt fractions using PySAM. + debt fractions using PySAM. - `./example_notebooks` Example Jupyter notebooks showing how to perform various operations. - `./tests` Tests for code in this repository. -## Citing this package +## Citing this Package + Mirletz, Brian, Bannister, Michael, Vimmerstedt, Laura, Stright, Dana, and Heine, Matthew. "ATB-calc (Annual Technology Baseline Calculators) [SWR-23-60]." Computer software. August 02, 2023. -https://github.com/NREL/ATB-calc. https://doi.org/10.11578/dc.20230914.2. \ No newline at end of file +https://github.com/NREL/ATB-calc. https://doi.org/10.11578/dc.20230914.2. + +## Code Formatting + +This project uses the [Black](https://black.readthedocs.io/en/stable/index.html) code formatting VS Code [plugin](https://marketplace.visualstudio.com/items?itemName=ms-python.black-formatter). Please install and run the plugin before making commits or a pull request. diff --git a/debt_fraction_calculator/debt_fraction_calc.py b/debt_fraction_calculator/debt_fraction_calc.py index ccd4a47..7b35f8f 100644 --- a/debt_fraction_calculator/debt_fraction_calc.py +++ b/debt_fraction_calculator/debt_fraction_calc.py @@ -16,31 +16,41 @@ import PySAM.Levpartflip as levpartflip from lcoe_calculator.extractor import Extractor -from lcoe_calculator.config import YEARS, END_YEAR, FINANCIAL_CASES, CrpChoiceType, PTC_PLUS_ITC_CASE_PVB +from lcoe_calculator.config import ( + YEARS, + END_YEAR, + FINANCIAL_CASES, + CrpChoiceType, + PTC_PLUS_ITC_CASE_PVB, +) from lcoe_calculator.tech_processors import LCOE_TECHS import lcoe_calculator.tech_processors from lcoe_calculator.base_processor import TechProcessor from lcoe_calculator.macrs import MACRS_6, MACRS_16, MACRS_21 -InputVals = TypedDict('InputVals', { - 'CF': float, # Capacity factor (%, 0-1) - 'OCC': float, # Overnight capital cost ($/kW) - 'CFC': float, # Construction financing cost ($/kW) - 'Fixed O&M': float, # Fixed operations and maintenance ($/kW-yr) - 'Variable O&M': float, # Variable operations and maintenance ($/Mwh) - 'DSCR': float, # Debt service coverage ratio (unitless, typically 1-1.5) - 'Rate of Return on Equity Nominal': float, # Internal rate of return (%, 0-1) - 'Tax Rate (Federal and State)': float, # (%, 0-1) - 'Inflation Rate': float, # (%, 0-1) - 'Fuel': float, # Fuel cost ($/MMBtu) - optional - 'Heat Rate': float, # (MMBtu/MWh) - optional - 'Interest Rate Nominal': float, # (%, 0-1) - 'Calculated Rate of Return on Equity Real': float, # (%, 0-1) - 'ITC': float, # Investment tax Credit, federal (%, 0-1) - 'PTC': float, # Production tax credit, federal ($/MWh) - 'MACRS': List[float], # Depreciation schedule from lcoe_calculator.macrs -}, total=False) +InputVals = TypedDict( + "InputVals", + { + "CF": float, # Capacity factor (%, 0-1) + "OCC": float, # Overnight capital cost ($/kW) + "CFC": float, # Construction financing cost ($/kW) + "Fixed O&M": float, # Fixed operations and maintenance ($/kW-yr) + "Variable O&M": float, # Variable operations and maintenance ($/Mwh) + "DSCR": float, # Debt service coverage ratio (unitless, typically 1-1.5) + "Rate of Return on Equity Nominal": float, # Internal rate of return (%, 0-1) + "Tax Rate (Federal and State)": float, # (%, 0-1) + "Inflation Rate": float, # (%, 0-1) + "Fuel": float, # Fuel cost ($/MMBtu) - optional + "Heat Rate": float, # (MMBtu/MWh) - optional + "Interest Rate Nominal": float, # (%, 0-1) + "Calculated Rate of Return on Equity Real": float, # (%, 0-1) + "ITC": float, # Investment tax Credit, federal (%, 0-1) + "PTC": float, # Production tax credit, federal ($/MWh) + "MACRS": List[float], # Depreciation schedule from lcoe_calculator.macrs + }, + total=False, +) def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: @@ -56,9 +66,11 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: # Values required for computation. Set to pysam using model.value() calls below analysis_period = 20 - ac_capacity = 1000 # kW + ac_capacity = 1000 # kW capacity_factor = input_vals["CF"] - gen = [capacity_factor * ac_capacity] * 8760 # Distribute evenly throughout the year + gen = [ + capacity_factor * ac_capacity + ] * 8760 # Distribute evenly throughout the year capex = input_vals["OCC"] con_fin_costs = input_vals["CFC"] @@ -73,7 +85,7 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: tax_state = 0 inflation = input_vals["Inflation Rate"] * 100 - degradation = 0.0 # ATB presents average capactity factors, so zero out degradation + degradation = 0.0 # ATB presents average capactity factors, so zero out degradation # Setting PySAM variables. See https://nrel-pysam.readthedocs.io/en/main/modules/Levpartflip.html for docs model.value("analysis_period", analysis_period) @@ -83,26 +95,35 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: model.value("total_installed_cost", initial_investment) ## Single Owner will apply the O&M cost to each year, so no need to multiply by analysis period - model.value("om_capacity", [o_and_m] ) + model.value("om_capacity", [o_and_m]) model.value("om_production", [v_o_and_m]) - if 'Fuel' in input_vals: - model.value("om_fuel_cost", [input_vals['Fuel']]) - if 'Heat Rate' in input_vals: - model.value("system_heat_rate", input_vals['Heat Rate']) + if "Fuel" in input_vals: + model.value("om_fuel_cost", [input_vals["Fuel"]]) + if "Heat Rate" in input_vals: + model.value("system_heat_rate", input_vals["Heat Rate"]) # Specify length 1 so degradation is applied each year. # An array of 0.7 len(analysis_period) assumes degradation the first year, but not afterwards model.value("degradation", [degradation]) - model.value("system_use_lifetime_output", 0) # Do degradation in the financial model + model.value( + "system_use_lifetime_output", 0 + ) # Do degradation in the financial model - model.value("debt_option", 1) # Use DSCR (alternative is to specify the debt fraction, which doesn't help) + model.value( + "debt_option", 1 + ) # Use DSCR (alternative is to specify the debt fraction, which doesn't help) model.value("dscr", dscr) model.value("inflation_rate", inflation) - model.value("term_int_rate", input_vals['Interest Rate Nominal'] * 100) - model.value("term_tenor", 18) # years - model.value("real_discount_rate", input_vals['Calculated Rate of Return on Equity Real'] * 100) - model.value("flip_target_percent", input_vals['Rate of Return on Equity Nominal'] * 100) ## "nominal equity rate" - model.value("flip_target_year", 10) # Assume flip occurs when PTC expires + model.value("term_int_rate", input_vals["Interest Rate Nominal"] * 100) + model.value("term_tenor", 18) # years + model.value( + "real_discount_rate", + input_vals["Calculated Rate of Return on Equity Real"] * 100, + ) + model.value( + "flip_target_percent", input_vals["Rate of Return on Equity Nominal"] * 100 + ) ## "nominal equity rate" + model.value("flip_target_year", 10) # Assume flip occurs when PTC expires model.value("ppa_escalation", 0.0) model.value("tax_investor_preflip_cash_percent", 90.0) @@ -132,9 +153,9 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: model.value("loan_moratorium", 0) model.value("construction_financing_cost", con_fin_total) model.value("itc_fed_percent", [input_vals["ITC"] * 100]) - model.value('itc_fed_percent_maxvalue', [1e38]) + model.value("itc_fed_percent_maxvalue", [1e38]) model.value("itc_sta_amount", [0]) - model.value("ptc_fed_amount", [input_vals["PTC"] / 1000]) # Convert $/MWh to $/kWh + model.value("ptc_fed_amount", [input_vals["PTC"] / 1000]) # Convert $/MWh to $/kWh # Production based incentive code to test treating the tax credits as available for debt service, currently unused model.value("pbi_fed_amount", [0]) @@ -165,8 +186,10 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: model.value("depr_itc_fed_sl_20", 1) model.value("depr_itc_fed_sl_20", 1) else: - raise ValueError('MACRS is expected to be one of MACRS_6, MACRS_16, or MACRS_21. ' - f'Unknown value provided: {input_vals["MACRS"]}') + raise ValueError( + "MACRS is expected to be one of MACRS_6, MACRS_16, or MACRS_21. " + f'Unknown value provided: {input_vals["MACRS"]}' + ) # Turn off unused depreciation features model.value("depr_alloc_custom_percent", 0) @@ -183,17 +206,19 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: model.value("depr_fedbas_method", 0) model.value("depr_stabas_method", 0) - model.value("ppa_soln_mode", 0) # Solve for PPA price given IRR - model.value("payment_option", 0) # Equal payments (standard amoritization) + model.value("ppa_soln_mode", 0) # Solve for PPA price given IRR + model.value("payment_option", 0) # Equal payments (standard amoritization) # Required for calculate PPA price. # Default is $0.045/kWh. However given the way we've set up gen, this will never be used - model.value('en_electricity_rates', 1 ) + model.value("en_electricity_rates", 1) model.execute() if debug: - print(f"LCOE: {model.Outputs.lcoe_real} cents/kWh") # multiply by 10 to get $ / MWh + print( + f"LCOE: {model.Outputs.lcoe_real} cents/kWh" + ) # multiply by 10 to get $ / MWh print(f"NPV: {model.Outputs.cf_project_return_aftertax_npv}") print() print(f"IRR in target year: {model.Outputs.flip_target_irr}") @@ -215,22 +240,30 @@ def calculate_debt_fraction(input_vals: InputVals, debug=False) -> float: tech_names = [Tech.__name__ for Tech in LCOE_TECHS] + @click.command -@click.argument('data_workbook_filename', type=click.Path(exists=True)) -@click.argument('output_filename', type=click.Path(exists=False)) -@click.option('-t', '--tech', type=click.Choice(tech_names), - help="Name of technology to calculate debt fraction for. Use all techs if none are " - "specified. Only technologies with an LCOE may be processed.") -@click.option('-d', '--debug', is_flag=True, default=False, help="Print debug data." ) -def calculate_all_debt_fractions(data_workbook_filename: str, output_filename: str, tech: str|None, - debug: bool): +@click.argument("data_workbook_filename", type=click.Path(exists=True)) +@click.argument("output_filename", type=click.Path(exists=False)) +@click.option( + "-t", + "--tech", + type=click.Choice(tech_names), + help="Name of technology to calculate debt fraction for. Use all techs if none are " + "specified. Only technologies with an LCOE may be processed.", +) +@click.option("-d", "--debug", is_flag=True, default=False, help="Print debug data.") +def calculate_all_debt_fractions( + data_workbook_filename: str, output_filename: str, tech: str | None, debug: bool +): """ Calculate debt fractions for one or more technologies, and all financial cases and years. DATA_WORKBOOK_FILENAME - Path and name of ATB data workbook XLXS file. OUTPUT_FILENAME - File to save calculated debt fractions to. Should end with .csv """ - tech_map: Dict[str, Type[TechProcessor]] = {tech.__name__: tech for tech in LCOE_TECHS} + tech_map: Dict[str, Type[TechProcessor]] = { + tech.__name__: tech for tech in LCOE_TECHS + } techs = LCOE_TECHS if tech is None else [tech_map[tech]] df_itc, df_ptc = Extractor.get_tax_credits_sheet(data_workbook_filename) @@ -245,63 +278,102 @@ def calculate_all_debt_fractions(data_workbook_filename: str, output_filename: s cols = ["Technology", "Case"] + [str(year) for year in tech_years] for fin_case in FINANCIAL_CASES: - click.echo(f"Processing tech {Tech.tech_name} and financial case {fin_case}") - debt_fracs = [Tech.tech_name, fin_case] # First two columns are metadata - - proc = Tech(data_workbook_filename, crp=crp, case=fin_case, tcc=PTC_PLUS_ITC_CASE_PVB) + click.echo( + f"Processing tech {Tech.tech_name} and financial case {fin_case}" + ) + debt_fracs = [Tech.tech_name, fin_case] # First two columns are metadata + + proc = Tech( + data_workbook_filename, + crp=crp, + case=fin_case, + tcc=PTC_PLUS_ITC_CASE_PVB, + ) proc.run() d = proc.flat # Values that are specific to the representative tech detail detail_vals = d[ - (d.DisplayName == Tech.default_tech_detail) & (d.Case == fin_case) - & (d.Scenario == 'Moderate') & (d.CRPYears == 20) & - ( - (d.Parameter == 'Fixed O&M') | (d.Parameter == 'Variable O&M') - | (d.Parameter == 'OCC') | (d.Parameter == 'CFC') | (d.Parameter == 'CF') - | (d.Parameter == 'Heat Rate') | (d.Parameter == 'Fuel') + (d.DisplayName == Tech.default_tech_detail) + & (d.Case == fin_case) + & (d.Scenario == "Moderate") + & (d.CRPYears == 20) + & ( + (d.Parameter == "Fixed O&M") + | (d.Parameter == "Variable O&M") + | (d.Parameter == "OCC") + | (d.Parameter == "CFC") + | (d.Parameter == "CF") + | (d.Parameter == "Heat Rate") + | (d.Parameter == "Fuel") ) ] # Values that apply to entire technology tech_vals = d[ - (d.Technology == Tech.tech_name) & (d.CRPYears == 20) & (d.Case == fin_case) & - ( - (d.Parameter == 'Inflation Rate') - | (d.Parameter == 'Tax Rate (Federal and State)') - | (d.Parameter == 'Calculated Rate of Return on Equity Real') - | (d.Parameter == 'Rate of Return on Equity Nominal') - | (d.Parameter == 'Interest Rate Nominal') + (d.Technology == Tech.tech_name) + & (d.CRPYears == 20) + & (d.Case == fin_case) + & ( + (d.Parameter == "Inflation Rate") + | (d.Parameter == "Tax Rate (Federal and State)") + | (d.Parameter == "Calculated Rate of Return on Equity Real") + | (d.Parameter == "Rate of Return on Equity Nominal") + | (d.Parameter == "Interest Rate Nominal") ) ] for year in tech_years: if debug: - click.echo(f"Processing tech {Tech.tech_name}, financial case {fin_case}, " - f"and year {year}") + click.echo( + f"Processing tech {Tech.tech_name}, financial case {fin_case}, " + f"and year {year}" + ) if not year in detail_vals or not year in tech_vals: debt_fracs.append(None) continue - input_vals = detail_vals.set_index('Parameter')[year].to_dict() - gen_vals = tech_vals.set_index('Parameter')[year].to_dict() + input_vals = detail_vals.set_index("Parameter")[year].to_dict() + gen_vals = tech_vals.set_index("Parameter")[year].to_dict() # Tax credits - assumes each tech has one PTC or one ITC - if Tech.has_tax_credit and fin_case == 'Market': + if Tech.has_tax_credit and fin_case == "Market": name = str(Tech.sheet_name) if Tech.wacc_name: name = Tech.wacc_name if Tech.sheet_name == "Utility-Scale PV-Plus-Battery": - if proc.tax_credit_case is PTC_PLUS_ITC_CASE_PVB and year > 2022: - ncf = proc.df_ncf.loc[Tech.default_tech_detail + '/Moderate'][year] - pvcf = proc.df_pvcf.loc[Tech.default_tech_detail + '/Moderate'][year] - - batt_occ_percent = proc.df_batt_cost * proc.CO_LOCATION_SAVINGS / proc.df_occ - - input_vals["PTC"] = df_ptc.loc[name][year] * min(ncf / pvcf, 1.0) - input_vals["ITC"] = df_itc.loc[name][year] * batt_occ_percent.loc[Tech.default_tech_detail + '/Moderate'][year] + if ( + proc.tax_credit_case is PTC_PLUS_ITC_CASE_PVB + and year > 2022 + ): + if Tech.default_tech_detail is None: + raise AttributeError( + "Tech.default_tech_detail must be set for " + ) + ncf = proc.df_ncf.loc[ + Tech.default_tech_detail + "/Moderate" + ][year] + pvcf = proc.df_pvcf.loc[ + Tech.default_tech_detail + "/Moderate" + ][year] + + batt_occ_percent = ( + proc.df_batt_cost + * proc.CO_LOCATION_SAVINGS + / proc.df_occ + ) + + input_vals["PTC"] = df_ptc.loc[name][year] * min( + ncf / pvcf, 1.0 + ) + input_vals["ITC"] = ( + df_itc.loc[name][year] + * batt_occ_percent.loc[ + Tech.default_tech_detail + "/Moderate" + ][year] + ) else: input_vals["PTC"] = 0 input_vals["ITC"] = df_itc.loc[name][year] @@ -319,16 +391,16 @@ def calculate_all_debt_fractions(data_workbook_filename: str, output_filename: s input_vals.update(gen_vals) - #Calculate debt fraction using PySAM + # Calculate debt fraction using PySAM debt_frac = calculate_debt_fraction(input_vals, debug) debt_frac /= 100.0 debt_fracs.append(debt_frac) debt_frac_dict[proc.tech_name + fin_case] = debt_fracs - debt_frac_df = pd.DataFrame.from_dict(debt_frac_dict, orient='index', columns=cols) + debt_frac_df = pd.DataFrame.from_dict(debt_frac_dict, orient="index", columns=cols) debt_frac_df.to_csv(output_filename) if __name__ == "__main__": - calculate_all_debt_fractions() # pylint: disable=no-value-for-parameter + calculate_all_debt_fractions() # pylint: disable=no-value-for-parameter diff --git a/lcoe_calculator/abstract_extractor.py b/lcoe_calculator/abstract_extractor.py index 4bb2db2..da7d4c1 100644 --- a/lcoe_calculator/abstract_extractor.py +++ b/lcoe_calculator/abstract_extractor.py @@ -17,8 +17,15 @@ class AbstractExtractor(ABC): """ @abstractmethod - def __init__(self, data_workbook_fname: str, sheet_name: str, case: str, crp: CrpChoiceType, - scenarios: List[str], base_year: int): + def __init__( + self, + data_workbook_fname: str, + sheet_name: str, + case: str, + crp: CrpChoiceType, + scenarios: List[str], + base_year: int, + ): """ @param data_workbook_fname - file name of data workbook @param sheet_name - name of sheet to process @@ -29,8 +36,9 @@ def __init__(self, data_workbook_fname: str, sheet_name: str, case: str, crp: Cr """ @abstractmethod - def get_metric_values(self, metric: str, num_tds: int, split_metrics: bool = False)\ - -> pd.DataFrame: + def get_metric_values( + self, metric: str, num_tds: int, split_metrics: bool = False + ) -> pd.DataFrame: """ Grab metric values table. @@ -42,7 +50,7 @@ def get_metric_values(self, metric: str, num_tds: int, split_metrics: bool = Fal @abstractmethod def get_tax_credits(self) -> pd.DataFrame: - """ Get tax credit """ + """Get tax credit""" @abstractmethod def get_cff(self, cff_name: str, rows: int) -> pd.DataFrame: @@ -64,7 +72,9 @@ def get_fin_assump(self) -> pd.DataFrame: """ @abstractmethod - def get_wacc(self, tech_name: str | None = None) -> Tuple[pd.DataFrame, pd.DataFrame]: + def get_wacc( + self, tech_name: str | None = None + ) -> Tuple[pd.DataFrame, pd.DataFrame]: """ Extract values for tech and case from WACC sheet. diff --git a/lcoe_calculator/base_processor.py b/lcoe_calculator/base_processor.py index 32b467c..ca643d2 100644 --- a/lcoe_calculator/base_processor.py +++ b/lcoe_calculator/base_processor.py @@ -15,8 +15,19 @@ from .macrs import MACRS_6 from .extractor import Extractor from .abstract_extractor import AbstractExtractor -from .config import FINANCIAL_CASES, END_YEAR, TECH_DETAIL_SCENARIO_COL, MARKET_FIN_CASE, CRP_CHOICES,\ - SCENARIOS, LCOE_SS_NAME, CAPEX_SS_NAME, CFF_SS_NAME, CrpChoiceType, BASE_YEAR +from .config import ( + FINANCIAL_CASES, + END_YEAR, + TECH_DETAIL_SCENARIO_COL, + MARKET_FIN_CASE, + CRP_CHOICES, + SCENARIOS, + LCOE_SS_NAME, + CAPEX_SS_NAME, + CFF_SS_NAME, + CrpChoiceType, + BASE_YEAR, +) class TechProcessor(ABC): @@ -39,12 +50,12 @@ class TechProcessor(ABC): @property @abstractmethod def sheet_name(self) -> str: - """ Name of the sheet in the excel data workbook """ + """Name of the sheet in the excel data workbook""" @property @abstractmethod def tech_name(self) -> str: - """ Name of tech for flat file""" + """Name of tech for flat file""" # For a consistent depreciation schedule, use one of the lists from the # macrs.py file as shown below. More complex schedules can be defined by @@ -55,12 +66,12 @@ def tech_name(self) -> str: # ------------ All other attributes have defaults ------------------------- # Metrics to load from SS. Format: (header in SS, object attribute name) metrics: List[Tuple[str, str]] = [ - ('Net Capacity Factor (%)', 'df_ncf'), - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - (CFF_SS_NAME, 'df_cff'), + ("Net Capacity Factor (%)", "df_ncf"), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + (CFF_SS_NAME, "df_cff"), ] tech_life = 30 # Tech lifespan in years @@ -71,40 +82,46 @@ def tech_name(self) -> str: has_tax_credit = True # Does the tech have tax credits in the workbook has_fin_assump = True # Does the tech have financial assumptions in the workbook - wacc_name: Optional[str] = None # Name of tech to look for on WACC sheet, use sheet name if None + wacc_name: Optional[str] = ( + None # Name of tech to look for on WACC sheet, use sheet name if None + ) has_wacc = True # If True, pull values from WACC sheet. - has_capex = True # If True, calculate CAPEX + has_capex = True # If True, calculate CAPEX has_lcoe = True # If True, calculate CRF, PFF, & LCOE. - split_metrics = False # Indicates 3 empty rows in tech detail metrics, e.g. hydropower + split_metrics = ( + False # Indicates 3 empty rows in tech detail metrics, e.g. hydropower + ) # Attributes to export in flat file, format: (attr name in class, value for # flat file). Any attributes that are None are silently ignored. Financial # assumptions values are added automatically. # See https://atb.nrel.gov/electricity/2023/acronyms or below for acronym definitions flat_attrs: List[Tuple[str, str]] = [ - ('df_ncf', 'CF'), - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_lcoe', 'LCOE'), - ('df_capex', 'CAPEX'), + ("df_ncf", "CF"), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_lcoe", "LCOE"), + ("df_capex", "CAPEX"), ] # Variables used by the debt fraction calculator. Should be filled out for any tech # where self.has_lcoe == True. default_tech_detail: Optional[str] = None - dscr: Optional[float] = None # Debt service coverage ratio (unitless, typically 1-1.5) + dscr: Optional[float] = ( + None # Debt service coverage ratio (unitless, typically 1-1.5) + ) def __init__( self, data_workbook_fname: str, case: str = MARKET_FIN_CASE, crp: CrpChoiceType = 30, - tcc : Optional[str] = None, - extractor: Type[AbstractExtractor] = Extractor + tcc: Optional[str] = None, + extractor: Type[AbstractExtractor] = Extractor, ): """ @param data_workbook_fname - name of workbook @@ -113,22 +130,24 @@ def __init__( @param tcc - tax credit case: 'ITC only' or 'PV PTC and Battery ITC' Only required for the PV plus battery technology. @param extractor - Extractor class to use to obtain source data. """ - assert case in FINANCIAL_CASES, (f'Financial case must be one of {FINANCIAL_CASES},' - f' received {case}') - assert crp in CRP_CHOICES, (f'Financial case must be one of {CRP_CHOICES},' - f' received {crp}') - assert isinstance(self.scenarios, list), 'self.scenarios must be a list' + assert case in FINANCIAL_CASES, ( + f"Financial case must be one of {FINANCIAL_CASES}," f" received {case}" + ) + assert crp in CRP_CHOICES, ( + f"Financial case must be one of {CRP_CHOICES}," f" received {crp}" + ) + assert isinstance(self.scenarios, list), "self.scenarios must be a list" if self.has_lcoe: if self.default_tech_detail is None: - raise ValueError('default_tech_detail must be set if has_lcoe is True.') + raise ValueError("default_tech_detail must be set if has_lcoe is True.") if self.dscr is None: - raise ValueError('dscr must be set if has_lcoe is True.') + raise ValueError("dscr must be set if has_lcoe is True.") self._data_workbook_fname = data_workbook_fname self._case = case self._requested_crp = crp - self._crp_years = self.tech_life if crp == 'TechLife' else crp + self._crp_years = self.tech_life if crp == "TechLife" else crp self._tech_years = range(self.base_year, END_YEAR + 1, 1) self.tax_credit_case = tcc @@ -142,9 +161,9 @@ def __init__( self.df_tc = None # Tax credits (varies) self.df_wacc = None # WACC table (varies) self.df_just_wacc = None # Last six rows of WACC table - self.df_hrp = None # Heat Rate Penalty (% change), retrofits only - self.df_nop = None # Net Output Penalty (% change), retrofits only - self.df_pvcf = None # PV-only capacity factor (%), PV-plus-battery only + self.df_hrp = None # Heat Rate Penalty (% change), retrofits only + self.df_nop = None # Net Output Penalty (% change), retrofits only + self.df_pvcf = None # PV-only capacity factor (%), PV-plus-battery only # These data frames are calculated and populated by object methods self.df_aep = None # Annual energy production (kWh/kW) @@ -158,21 +177,22 @@ def __init__( self._extractor = self._extract_data() def run(self): - """ Run all calculations for CAPEX and LCOE """ + """Run all calculations for CAPEX and LCOE""" if self.has_capex: self.df_cfc = self._calc_con_fin_cost() self.df_capex = self._calc_capex() - assert not self.df_capex.isnull().any().any(),\ - f'Error in calculated CAPEX, found missing values: {self.df_capex}' + assert ( + not self.df_capex.isnull().any().any() + ), f"Error in calculated CAPEX, found missing values: {self.df_capex}" if self.has_lcoe and self.has_wacc: self.df_aep = self._calc_aep() self.df_crf = self._calc_crf() self.df_pff = self._calc_pff() self.df_lcoe = self._calc_lcoe() - assert not self.df_lcoe.isnull().any().any(),\ - f'Error in calculated LCOE, found missing values: {self.df_lcoe}' - + assert ( + not self.df_lcoe.isnull().any().any() + ), f"Error in calculated LCOE, found missing values: {self.df_lcoe}" @property def flat(self) -> pd.DataFrame: @@ -187,7 +207,7 @@ def flat(self) -> pd.DataFrame: df_flat = pd.DataFrame() if self.df_wacc is None else self._flat_fin_assump() case = self._case.upper() - if case == 'MARKET': + if case == "MARKET": case = MARKET_FIN_CASE for attr, parameter in self.flat_attrs: @@ -195,20 +215,28 @@ def flat(self) -> pd.DataFrame: df = df.reset_index() old_cols = df.columns - df[['DisplayName', 'Scenario']] = df[TECH_DETAIL_SCENARIO_COL].str\ - .rsplit('/', n=1, expand=True) + df[["DisplayName", "Scenario"]] = df[TECH_DETAIL_SCENARIO_COL].str.rsplit( + "/", n=1, expand=True + ) df.DisplayName = df.DisplayName.str.strip() df.Scenario = df.Scenario.str.strip() - df['Parameter'] = parameter + df["Parameter"] = parameter df_flat = pd.concat([df_flat, df]) - df_flat['Technology'] = self.tech_name - df_flat['Case'] = case - df_flat['CRPYears'] = self._crp_years - df_flat['TaxCreditCase'] = self._get_tax_credit_case() - - new_cols = ['Parameter', 'Case', 'TaxCreditCase', 'CRPYears', 'Technology', 'DisplayName', - 'Scenario'] + list(old_cols) + df_flat["Technology"] = self.tech_name + df_flat["Case"] = case + df_flat["CRPYears"] = self._crp_years + df_flat["TaxCreditCase"] = self._get_tax_credit_case() + + new_cols = [ + "Parameter", + "Case", + "TaxCreditCase", + "CRPYears", + "Technology", + "DisplayName", + "Scenario", + ] + list(old_cols) df_flat = df_flat[new_cols] df_flat = df_flat.drop(TECH_DETAIL_SCENARIO_COL, axis=1).reset_index(drop=True) @@ -235,23 +263,27 @@ def test_lcoe(self): if there is a discrepancy. """ if not self.has_lcoe: - print(f'LCOE is not calculated for {self.sheet_name}, skipping test.') + print(f"LCOE is not calculated for {self.sheet_name}, skipping test.") return - assert self.df_lcoe is not None, 'Please run `run()` first to calculate LCOE.' - - self.ss_lcoe = self._extractor.get_metric_values(LCOE_SS_NAME, self.num_tds, - self.split_metrics) - - assert not self.df_lcoe.isnull().any().any(),\ - f'Error in calculated LCOE, found missing values: {self.df_lcoe}' - assert not self.ss_lcoe.isnull().any().any(),\ - f'Error in LCOE from workbook, found missing values: {self.ss_lcoe}' - - if np.allclose(np.array(self.df_lcoe, dtype=float), - np.array(self.ss_lcoe, dtype=float)): - print('Calculated LCOE matches LCOE from workbook') + assert self.df_lcoe is not None, "Please run `run()` first to calculate LCOE." + + self.ss_lcoe = self._extractor.get_metric_values( + LCOE_SS_NAME, self.num_tds, self.split_metrics + ) + + assert ( + not self.df_lcoe.isnull().any().any() + ), f"Error in calculated LCOE, found missing values: {self.df_lcoe}" + assert ( + not self.ss_lcoe.isnull().any().any() + ), f"Error in LCOE from workbook, found missing values: {self.ss_lcoe}" + + if np.allclose( + np.array(self.df_lcoe, dtype=float), np.array(self.ss_lcoe, dtype=float) + ): + print("Calculated LCOE matches LCOE from workbook") else: - msg = f'Calculated LCOE doesn\'t match LCOE from workbook for {self.sheet_name}' + msg = f"Calculated LCOE doesn't match LCOE from workbook for {self.sheet_name}" print(msg) print("Workbook LCOE:") print(self.ss_lcoe) @@ -265,22 +297,26 @@ def test_capex(self): if there is a discrepancy. """ if not self.has_capex: - print(f'CAPEX is not calculated for {self.sheet_name}, skipping test.') + print(f"CAPEX is not calculated for {self.sheet_name}, skipping test.") return - assert self.df_capex is not None, 'Please run `run()` first to calculate CAPEX.' - - self.ss_capex = self._extractor.get_metric_values(CAPEX_SS_NAME, self.num_tds, - self.split_metrics) - - assert not self.df_capex.isnull().any().any(),\ - f'Error in calculated CAPEX, found missing values: {self.df_capex}' - assert not self.ss_capex.isnull().any().any(),\ - f'Error in CAPEX from workbook, found missing values: {self.ss_capex}' - if np.allclose(np.array(self.df_capex, dtype=float), - np.array(self.ss_capex, dtype=float)): - print('Calculated CAPEX matches CAPEX from workbook') + assert self.df_capex is not None, "Please run `run()` first to calculate CAPEX." + + self.ss_capex = self._extractor.get_metric_values( + CAPEX_SS_NAME, self.num_tds, self.split_metrics + ) + + assert ( + not self.df_capex.isnull().any().any() + ), f"Error in calculated CAPEX, found missing values: {self.df_capex}" + assert ( + not self.ss_capex.isnull().any().any() + ), f"Error in CAPEX from workbook, found missing values: {self.ss_capex}" + if np.allclose( + np.array(self.df_capex, dtype=float), np.array(self.ss_capex, dtype=float) + ): + print("Calculated CAPEX matches CAPEX from workbook") else: - raise ValueError('Calculated CAPEX doesn\'t match CAPEX from workbook') + raise ValueError("Calculated CAPEX doesn't match CAPEX from workbook") def _flat_fin_assump(self): """ @@ -289,34 +325,37 @@ def _flat_fin_assump(self): @returns {pd.DataFrame} """ - assert self.df_wacc is not None, ('df_wacc must not be None to flatten ' - 'financial assumptions.') + assert self.df_wacc is not None, ( + "df_wacc must not be None to flatten " "financial assumptions." + ) df = self.df_wacc.copy() # Add CRF and FCR if self.has_tax_credit and self.df_pff is not None: for scenario in self.scenarios: - wacc = df.loc[f'WACC Real - {scenario}'] - pff = self.df_pff.loc[f'PFF - {scenario}'] + wacc = df.loc[f"WACC Real - {scenario}"] + pff = self.df_pff.loc[f"PFF - {scenario}"] crf, fcr = self._calc_fcr(wacc, self._crp_years, pff, scenario) df = pd.concat([df, crf, fcr]) else: # No tax credit, just fill with * cols = df.columns - fcr = pd.DataFrame({c:['*'] for c in cols}, index=['FCR']) - crf = pd.DataFrame({c:['*'] for c in cols}, index=['CRF']) + fcr = pd.DataFrame({c: ["*"] for c in cols}, index=["FCR"]) + crf = pd.DataFrame({c: ["*"] for c in cols}, index=["CRF"]) df = pd.concat([df, crf, fcr]) # Explode index and clean up - df.index.rename('WACC', inplace=True) + df.index.rename("WACC", inplace=True) df = df.reset_index(drop=False) - df[['Parameter', 'Scenario']] = df.WACC.str.split(' - ', expand=True) - df.loc[df.Scenario.isnull(), 'Scenario'] = '*' - df.loc[df.Scenario == 'Nominal', 'Parameter'] = 'Interest During Construction - Nominal' - df.loc[df.Scenario == 'Nominal', 'Scenario'] = '*' - df['DisplayName'] = '*' - df['TaxCreditCase'] = self._get_tax_credit_case() + df[["Parameter", "Scenario"]] = df.WACC.str.split(" - ", expand=True) + df.loc[df.Scenario.isnull(), "Scenario"] = "*" + df.loc[df.Scenario == "Nominal", "Parameter"] = ( + "Interest During Construction - Nominal" + ) + df.loc[df.Scenario == "Nominal", "Scenario"] = "*" + df["DisplayName"] = "*" + df["TaxCreditCase"] = self._get_tax_credit_case() return df @@ -332,27 +371,36 @@ def _calc_fcr(wacc, crp, pff, scenario): @returns {pd.DataFrame, pd.DataFrame} - CRF and FCR """ - crf = wacc/(1 - 1/(1 + wacc)**crp) - fcr = crf*pff - crf.name = f'CRF - {scenario}' - fcr.name = f'FCR - {scenario}' + crf = wacc / (1 - 1 / (1 + wacc) ** crp) + fcr = crf * pff + crf.name = f"CRF - {scenario}" + fcr.name = f"FCR - {scenario}" crf = pd.DataFrame(crf).T fcr = pd.DataFrame(fcr).T return crf, fcr def _extract_data(self): - """ Pull all data from the workbook """ - crp_msg = self._requested_crp if self._requested_crp != 'TechLife' \ - else f'TechLife ({self.tech_life})' - - print(f'Loading data from {self.sheet_name}, for {self._case} and {crp_msg}') - extractor = self._ExtractorClass(self._data_workbook_fname, self.sheet_name, - self._case, self._requested_crp, self.scenarios, self.base_year) - - print('\tLoading metrics') + """Pull all data from the workbook""" + crp_msg = ( + self._requested_crp + if self._requested_crp != "TechLife" + else f"TechLife ({self.tech_life})" + ) + + print(f"Loading data from {self.sheet_name}, for {self._case} and {crp_msg}") + extractor = self._ExtractorClass( + self._data_workbook_fname, + self.sheet_name, + self._case, + self._requested_crp, + self.scenarios, + self.base_year, + ) + + print("\tLoading metrics") for metric, var_name in self.metrics: - if var_name == 'df_cff': + if var_name == "df_cff": # Grab DF index from another value to use in full CFF DF index = getattr(self, self.metrics[0][1]).index self.df_cff = self.load_cff(extractor, metric, index) @@ -365,20 +413,21 @@ def _extract_data(self): self.df_tc = extractor.get_tax_credits() # Pull financial assumptions from small table at top of tech sheet - print('\tLoading assumptions') + print("\tLoading assumptions") if self.has_fin_assump: self.df_fin = extractor.get_fin_assump() if self.has_wacc: - print('\tLoading WACC data') + print("\tLoading WACC data") self.df_wacc, self.df_just_wacc = extractor.get_wacc(self.wacc_name) - print('\tDone loading data') + print("\tDone loading data") return extractor @classmethod - def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, - return_short_df=False) -> pd.DataFrame: + def load_cff( + cls, extractor: Extractor, cff_name: str, index: pd.Index, return_short_df=False + ) -> pd.DataFrame: """ Load CFF data from workbook and duplicate for all tech details. This method is a little weird due to testing needs. @@ -390,9 +439,10 @@ def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, @returns - CFF data frame """ df_cff = extractor.get_cff(cff_name, len(cls.scenarios)) - assert len(df_cff) == len(cls.scenarios),\ - (f'Wrong number of CFF rows found. Expected {len(cls.scenarios)}, ' - f'get {len(df_cff)}.') + assert len(df_cff) == len(cls.scenarios), ( + f"Wrong number of CFF rows found. Expected {len(cls.scenarios)}, " + f"get {len(df_cff)}." + ) if return_short_df: return df_cff @@ -406,13 +456,16 @@ def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, return full_df_cff def _calc_aep(self): - assert self.df_ncf is not None, 'NCF must to loaded to calculate AEP' + assert self.df_ncf is not None, "NCF must to loaded to calculate AEP" df_aep = self.df_ncf * 8760 return df_aep def _calc_capex(self): - assert self.df_cff is not None and self.df_occ is not None and\ - self.df_gcc is not None, 'CFF, OCC, and GCC must to loaded to calculate CAPEX' + assert ( + self.df_cff is not None + and self.df_occ is not None + and self.df_gcc is not None + ), "CFF, OCC, and GCC must to loaded to calculate CAPEX" df_capex = self.df_cff * (self.df_occ + self.df_gcc) df_capex = df_capex.copy() return df_capex @@ -423,13 +476,15 @@ def _calc_con_fin_cost(self): return df_cfc def _calc_crf(self): - df_crf = self.df_just_wacc/(1-(1/(1+self.df_just_wacc))**self.crp) + df_crf = self.df_just_wacc / (1 - (1 / (1 + self.df_just_wacc)) ** self.crp) # Relabel WACC index as CRF df_crf = df_crf.reset_index() - df_crf['WACC Type'] = df_crf['WACC Type'].apply(lambda x: 'Capital Recovery Factor (CRF)'+x[4:]) - df_crf = df_crf.loc[df_crf['WACC Type'].str.contains('Real')] - df_crf = df_crf.set_index('WACC Type') + df_crf["WACC Type"] = df_crf["WACC Type"].apply( + lambda x: "Capital Recovery Factor (CRF)" + x[4:] + ) + df_crf = df_crf.loc[df_crf["WACC Type"].str.contains("Real")] + df_crf = df_crf.set_index("WACC Type") return df_crf @@ -440,20 +495,22 @@ def crp(self) -> float: @returns: CRP """ - raw_crp = self.df_fin.loc['Capital Recovery Period (Years)', 'Value'] + raw_crp = self.df_fin.loc["Capital Recovery Period (Years)", "Value"] try: crp = float(raw_crp) except ValueError as err: - msg = f'Error converting CRP value ({raw_crp}) to a float: {err}.' - print(f'{msg} self.df_fin is:') + msg = f"Error converting CRP value ({raw_crp}) to a float: {err}." + print(f"{msg} self.df_fin is:") print(self.df_fin) raise ValueError(msg) from err - assert not np.isnan(crp), f'CRP must be a number, got "{crp}", type is "{type(crp)}"' + assert not np.isnan( + crp + ), f'CRP must be a number, got "{crp}", type is "{type(crp)}"' return crp - def _calc_itc(self, itc_type=''): + def _calc_itc(self, itc_type=""): """ Calculate ITC if used @@ -461,9 +518,11 @@ def _calc_itc(self, itc_type=''): @returns {np.ndarray|int} - array of ITC values or 0 """ if self.has_tax_credit: - itc_index = f'ITC Schedule{itc_type}/*' - assert itc_index in self.df_tc.index, ('ITC schedule not found in ' - f'tax credit data. Looking for "{itc_index}" in:\n{self.df_tc}') + itc_index = f"ITC Schedule{itc_type}/*" + assert itc_index in self.df_tc.index, ( + "ITC schedule not found in " + f'tax credit data. Looking for "{itc_index}" in:\n{self.df_tc}' + ) df_itc_schedule = self.df_tc.loc[itc_index] itc_schedule = df_itc_schedule.values else: @@ -471,15 +530,15 @@ def _calc_itc(self, itc_type=''): return itc_schedule - def _calc_pff(self, itc_type=''): + def _calc_pff(self, itc_type=""): """ Calculate PFF @param {str} itc_type - type of ITC to search for (used for utility PV + batt) @returns {pd.DataFrame} - dataframe of PFF """ - df_tax_rate = self.df_wacc.loc['Tax Rate (Federal and State)'] - inflation = self.df_wacc.loc['Inflation Rate'] + df_tax_rate = self.df_wacc.loc["Tax Rate (Federal and State)"] + inflation = self.df_wacc.loc["Inflation Rate"] df_pvd = pd.DataFrame(columns=self._tech_years) for scenario in self.scenarios: @@ -491,14 +550,16 @@ def _calc_pff(self, itc_type=''): MACRS_schedule, inflation, scenario ) - df_pvd.loc['PVD - ' + scenario,year] = np.dot(MACRS_schedule, - df_depreciation_factor[year]) + df_pvd.loc["PVD - " + scenario, year] = np.dot( + MACRS_schedule, df_depreciation_factor[year] + ) itc_schedule = self._calc_itc(itc_type=itc_type) - df_pff = (1 - df_tax_rate.values*df_pvd*(1-itc_schedule/2) - - itc_schedule)/(1-df_tax_rate.values) - df_pff.index = [f'PFF - {scenario}' for scenario in self.scenarios] + df_pff = ( + 1 - df_tax_rate.values * df_pvd * (1 - itc_schedule / 2) - itc_schedule + ) / (1 - df_tax_rate.values) + df_pff.index = [f"PFF - {scenario}" for scenario in self.scenarios] return df_pff def _calc_dep_factor(self, MACRS_schedule, inflation, scenario): @@ -514,10 +575,12 @@ def _calc_dep_factor(self, MACRS_schedule, inflation, scenario): """ dep_years = len(MACRS_schedule) df_depreciation_factor = pd.DataFrame(columns=self._tech_years) - wacc_real = self.df_wacc.loc['WACC Real - ' + scenario] + wacc_real = self.df_wacc.loc["WACC Real - " + scenario] for dep_year in range(dep_years): - df_depreciation_factor.loc[dep_year+1] = 1/((1+wacc_real)*(1+inflation))**(dep_year+1) + df_depreciation_factor.loc[dep_year + 1] = 1 / ( + (1 + wacc_real) * (1 + inflation) + ) ** (dep_year + 1) return df_depreciation_factor @@ -529,12 +592,16 @@ def _calc_ptc(self): """ if self.has_tax_credit: df_tax_credit = self.df_tc.reset_index() - df_ptc = df_tax_credit.loc[df_tax_credit['Tax Credit'].str.contains('PTC/', na=False)] + df_ptc = df_tax_credit.loc[ + df_tax_credit["Tax Credit"].str.contains("PTC/", na=False) + ] - assert len(df_ptc) != 0, f'PTC data is missing for {self.sheet_name}' - assert len(df_ptc) == len(self.scenarios), f'Wrong amount of PTC data for{self.sheet_name}' + assert len(df_ptc) != 0, f"PTC data is missing for {self.sheet_name}" + assert len(df_ptc) == len( + self.scenarios + ), f"Wrong amount of PTC data for{self.sheet_name}" - df_ptc = pd.concat([df_ptc] * self.num_tds).set_index('Tax Credit') + df_ptc = pd.concat([df_ptc] * self.num_tds).set_index("Tax Credit") ptc = df_ptc.values else: ptc = 0 @@ -544,15 +611,17 @@ def _calc_ptc(self): def _calc_lcoe(self): ptc = self._calc_ptc() - assert len(self.df_crf) == len(self.scenarios),\ - (f'CRF has {len(self.df_crf)} rows ({self.df_crf.index}), but there ' - f'are {len(self.scenarios)} scenarios ({self.scenarios})') + assert len(self.df_crf) == len(self.scenarios), ( + f"CRF has {len(self.df_crf)} rows ({self.df_crf.index}), but there " + f"are {len(self.scenarios)} scenarios ({self.scenarios})" + ) x = self.df_crf.values * self.df_pff y = pd.concat([x] * self.num_tds) - df_lcoe = (1000 * (y.values * self.df_capex.values + self.df_fom)/ - self.df_aep.values) + df_lcoe = ( + 1000 * (y.values * self.df_capex.values + self.df_fom) / self.df_aep.values + ) df_lcoe = df_lcoe + self.df_vom.values - ptc return df_lcoe @@ -566,8 +635,9 @@ def _get_tax_credit_case(self): """ if not self.has_tax_credit: return "None" - assert len(self.df_tc) > 0, \ - (f'Setup df_tc with extractor.get_tax_credits() before calling this function!') + assert ( + len(self.df_tc) > 0 + ), f"Setup df_tc with extractor.get_tax_credits() before calling this function!" ptc = self._calc_ptc() itc = self._calc_itc() @@ -586,4 +656,4 @@ def _get_tax_credit_case(self): if itc_sum > 0: return "ITC" else: - return "None" \ No newline at end of file + return "None" diff --git a/lcoe_calculator/config.py b/lcoe_calculator/config.py index 2b0fea5..feecb09 100644 --- a/lcoe_calculator/config.py +++ b/lcoe_calculator/config.py @@ -12,29 +12,31 @@ # Years of data projected by ATB BASE_YEAR = 2022 END_YEAR = 2050 -YEARS = list(range(BASE_YEAR, END_YEAR + 1 , 1)) +YEARS = list(range(BASE_YEAR, END_YEAR + 1, 1)) # Financial cases -MARKET_FIN_CASE = 'Market' -R_AND_D_FIN_CASE = 'R&D' +MARKET_FIN_CASE = "Market" +R_AND_D_FIN_CASE = "R&D" FINANCIAL_CASES = [MARKET_FIN_CASE, R_AND_D_FIN_CASE] # Tax credit cases -ITC_ONLY_CASE = 'ITC only' -PTC_PLUS_ITC_CASE_PVB = 'PV PTC and Battery ITC' -TAX_CREDIT_CASES = {'Utility-Scale PV-Plus-Battery' : [ITC_ONLY_CASE, PTC_PLUS_ITC_CASE_PVB]} +ITC_ONLY_CASE = "ITC only" +PTC_PLUS_ITC_CASE_PVB = "PV PTC and Battery ITC" +TAX_CREDIT_CASES = { + "Utility-Scale PV-Plus-Battery": [ITC_ONLY_CASE, PTC_PLUS_ITC_CASE_PVB] +} # CRP choices and type hints -CrpChoiceType = Literal[20, 30, 'TechLife'] -CRP_CHOICES: List[CrpChoiceType] = [20, 30, 'TechLife'] +CrpChoiceType = Literal[20, 30, "TechLife"] +CRP_CHOICES: List[CrpChoiceType] = [20, 30, "TechLife"] # Technology advancement scenarios -SCENARIOS = ['Advanced', 'Moderate', 'Conservative'] +SCENARIOS = ["Advanced", "Moderate", "Conservative"] # Column name for combined tech detail name and scenario, aka Column K in the workbook -TECH_DETAIL_SCENARIO_COL = 'tech_detail-scenario' +TECH_DETAIL_SCENARIO_COL = "tech_detail-scenario" # Metric header names in ATB data workbook -LCOE_SS_NAME = 'Levelized Cost of Energy ($/MWh)' -CAPEX_SS_NAME = 'CAPEX ($/kW)' -CFF_SS_NAME = 'Construction Finance Factor' \ No newline at end of file +LCOE_SS_NAME = "Levelized Cost of Energy ($/MWh)" +CAPEX_SS_NAME = "CAPEX ($/kW)" +CFF_SS_NAME = "Construction Finance Factor" diff --git a/lcoe_calculator/extractor.py b/lcoe_calculator/extractor.py index 4bb7d20..19fc887 100644 --- a/lcoe_calculator/extractor.py +++ b/lcoe_calculator/extractor.py @@ -26,11 +26,19 @@ class Extractor(AbstractExtractor): """ Extract financial assumptions, metrics, and WACC from Excel data workbook. """ - wacc_sheet = 'WACC Calc' - tax_credits_sheet = 'Tax Credits' - def __init__(self, data_workbook_fname: str, sheet_name: str, case: str, crp: CrpChoiceType, - scenarios: List[str], base_year: int): + wacc_sheet = "WACC Calc" + tax_credits_sheet = "Tax Credits" + + def __init__( + self, + data_workbook_fname: str, + sheet_name: str, + case: str, + crp: CrpChoiceType, + scenarios: List[str], + base_year: int, + ): """ @param data_workbook_fname - file name of data workbook @param sheet_name - name of sheet to process @@ -49,21 +57,21 @@ def __init__(self, data_workbook_fname: str, sheet_name: str, case: str, crp: Cr # Open workbook, set fin case and CRP, and save wb = xw.Book(data_workbook_fname) - sheet = wb.sheets['Financial and CRP Inputs'] - sheet.range('B5').value = case - sheet.range('E5').value = crp + sheet = wb.sheets["Financial and CRP Inputs"] + sheet.range("B5").value = case + sheet.range("E5").value = crp wb.save() df = pd.read_excel(data_workbook_fname, sheet_name=sheet_name) df = df.reset_index() # Give columns numerical names - columns = {x:y for x,y in zip(df.columns,range(0,len(df.columns)))} + columns = {x: y for x, y in zip(df.columns, range(0, len(df.columns)))} df = df.rename(columns=columns) self._df = df # Grab tech values and header - tables_start_row, _ = self._find_cell(df, 'Future Projections') - tables_end_row, _ = self._find_cell(df, 'Data Sources for Default Inputs') + tables_start_row, _ = self._find_cell(df, "Future Projections") + tables_end_row, _ = self._find_cell(df, "Data Sources for Default Inputs") self._df_tech_header = df.loc[0:tables_start_row] self._df_tech_full = df.loc[tables_start_row:tables_end_row] @@ -81,48 +89,60 @@ def get_tax_credits_sheet(cls, data_workbook_fname): df_tc = df_tc.reset_index() # Give columns numerical names - columns = {x:y for x,y in zip(df_tc.columns,range(0,len(df_tc.columns)))} + columns = {x: y for x, y in zip(df_tc.columns, range(0, len(df_tc.columns)))} df_tc = df_tc.rename(columns=columns) - #First and last year locations in header + # First and last year locations in header fy_row, fy_col = cls._find_cell(df_tc, YEARS[0]) ly_row, ly_col = cls._find_cell(df_tc, YEARS[-1]) - assert fy_row == ly_row, 'First and last year headings were not found on the same row '+\ - 'on the tax credit sheet.' + assert fy_row == ly_row, ( + "First and last year headings were not found on the same row " + + "on the tax credit sheet." + ) # Figure out location of data - itc_row, itc_col = cls._find_cell(df_tc, 'ITC (%)') - ptc_row, ptc_col = cls._find_cell(df_tc, 'PTC ($/MWh)') - assert itc_col + 2 == fy_col, 'Expected first data column for ITC does not line up '+\ - 'with first year heading.' - assert ptc_col + 2 == fy_col, 'Expected first data column for PTC does not line up '+\ - 'with first year heading.' - assert itc_col == ptc_col, 'ITC and PTC marker text are not in the same column' + itc_row, itc_col = cls._find_cell(df_tc, "ITC (%)") + ptc_row, ptc_col = cls._find_cell(df_tc, "PTC ($/MWh)") + assert itc_col + 2 == fy_col, ( + "Expected first data column for ITC does not line up " + + "with first year heading." + ) + assert ptc_col + 2 == fy_col, ( + "Expected first data column for PTC does not line up " + + "with first year heading." + ) + assert itc_col == ptc_col, "ITC and PTC marker text are not in the same column" # Pull years from tax credit sheet years = list(df_tc.loc[fy_row, fy_col:ly_col].astype(int).values) - assert years == YEARS, f'Years in tax credit sheet ({years}) do not match ATB years ({YEARS})' + assert ( + years == YEARS + ), f"Years in tax credit sheet ({years}) do not match ATB years ({YEARS})" # Pull ITC and PTC values - df_itc = df_tc.loc[itc_row:ptc_row-2, itc_col+1:ly_col] - df_itc.columns = ['Technology'] + years + df_itc = df_tc.loc[itc_row : ptc_row - 2, itc_col + 1 : ly_col] + df_itc.columns = ["Technology"] + years df_itc.index = df_itc.Technology - df_itc.drop('Technology', axis=1, inplace=True) + df_itc.drop("Technology", axis=1, inplace=True) - df_ptc = df_tc.loc[ptc_row:, ptc_col+1:ly_col] - df_ptc.columns = ['Technology'] + years + df_ptc = df_tc.loc[ptc_row:, ptc_col + 1 : ly_col] + df_ptc.columns = ["Technology"] + years df_ptc.index = df_ptc.Technology - df_ptc.drop('Technology', axis=1, inplace=True) + df_ptc.drop("Technology", axis=1, inplace=True) df_ptc = df_ptc.dropna() - assert not df_itc.isnull().any().any(),\ - f'Error loading ITC. Found empty values: {df_itc}' - assert not df_ptc.isnull().any().any(),\ - f'Error loading PTC. Found empty values: {df_ptc}' + assert ( + not df_itc.isnull().any().any() + ), f"Error loading ITC. Found empty values: {df_itc}" + assert ( + not df_ptc.isnull().any().any() + ), f"Error loading PTC. Found empty values: {df_ptc}" return df_itc, df_ptc - def get_wacc(self, tech_name: str | None = None) -> Tuple[pd.DataFrame, pd.DataFrame]: + def get_wacc( + self, tech_name: str | None = None + ) -> Tuple[pd.DataFrame, pd.DataFrame]: """ Extract values for tech and case from WACC sheet. @@ -133,49 +153,59 @@ def get_wacc(self, tech_name: str | None = None) -> Tuple[pd.DataFrame, pd.DataF Real - {scenario}' """ df_wacc = pd.read_excel(self._data_workbook_fname, self.wacc_sheet) - case = 'Market Factors' if self._case == 'Market' else 'R&D' + case = "Market Factors" if self._case == "Market" else "R&D" tech_name = self.sheet_name if tech_name is None else tech_name - search = f'{tech_name} {case}' + search = f"{tech_name} {case}" count = (df_wacc == search).sum().sum() if count != 1: assert count != 0, f'Unable to find "{search}" on {self.wacc_sheet} sheet.' - assert count <= 1, f'"{search}" found more than once in {self.wacc_sheet} sheet.' + assert ( + count <= 1 + ), f'"{search}" found more than once in {self.wacc_sheet} sheet.' start_row, c = self._find_cell(df_wacc, search) - assert c == 'Unnamed: 0', f'WACC Calc tech search string ("{search}") found in wrong column' + assert ( + c == "Unnamed: 0" + ), f'WACC Calc tech search string ("{search}") found in wrong column' # Grab the rows, reset index and columns - df_wacc = df_wacc.iloc[start_row:start_row + NUM_WACC_PARMS + 1] - df_wacc = df_wacc.set_index('Unnamed: 1') + df_wacc = df_wacc.iloc[start_row : start_row + NUM_WACC_PARMS + 1] + df_wacc = df_wacc.set_index("Unnamed: 1") df_wacc.columns = pd.Index(df_wacc.iloc[0]) # Drop empty columns and first row w/ years - df_wacc = df_wacc.dropna(axis=1, how='any').drop(df_wacc.index[0]) - df_wacc.index.rename('WACC', inplace=True) + df_wacc = df_wacc.dropna(axis=1, how="any").drop(df_wacc.index[0]) + df_wacc.index.rename("WACC", inplace=True) df_wacc.columns = df_wacc.columns.astype(int) - df_wacc.columns.name = 'year' + df_wacc.columns.name = "year" df_just_wacc = df_wacc.iloc[-6:] - df_just_wacc.index.rename('WACC Type', inplace=True) + df_just_wacc.index.rename("WACC Type", inplace=True) idx = df_wacc.index - assert idx[0] == 'Inflation Rate' and idx[-1] == 'WACC Real - Conservative', \ - ('"Inflation Rate" should be the first row in the WACC table and ' - f'"WACC Real - Conservative" should be last, but "{idx[0]}" and ' - f'"{idx[-1]}" were found instead. Please check the data workbook ' - f'and NUM_WACC_PARAMS.') + assert idx[0] == "Inflation Rate" and idx[-1] == "WACC Real - Conservative", ( + '"Inflation Rate" should be the first row in the WACC table and ' + f'"WACC Real - Conservative" should be last, but "{idx[0]}" and ' + f'"{idx[-1]}" were found instead. Please check the data workbook ' + f"and NUM_WACC_PARAMS." + ) cols = df_wacc.columns - assert cols[0] == YEARS[0], f'WACC: First year should be {YEARS[0]}, got {cols[0]} instead' - assert cols[-1] == YEARS[-1], f'WACC: Last year should be {YEARS[-1]}, got {cols[-1]} instead' + assert ( + cols[0] == YEARS[0] + ), f"WACC: First year should be {YEARS[0]}, got {cols[0]} instead" + assert ( + cols[-1] == YEARS[-1] + ), f"WACC: Last year should be {YEARS[-1]}, got {cols[-1]} instead" if self.base_year != YEARS[0]: - df_wacc = df_wacc.loc[:, self.base_year:YEARS[-1]] - df_just_wacc = df_just_wacc.loc[:, self.base_year:YEARS[-1]] + df_wacc = df_wacc.loc[:, self.base_year : YEARS[-1]] + df_just_wacc = df_just_wacc.loc[:, self.base_year : YEARS[-1]] - assert not df_wacc.isnull().any().any(),\ - f'Error loading WACC for {tech_name}. Found empty values: {df_wacc}' + assert ( + not df_wacc.isnull().any().any() + ), f"Error loading WACC for {tech_name}. Found empty values: {df_wacc}" return df_wacc, df_just_wacc @@ -186,10 +216,10 @@ def get_fin_assump(self) -> pd.DataFrame: @returns financial assumption data """ - r1, c = self._find_cell(self._df, 'Financial Assumptions:') + r1, c = self._find_cell(self._df, "Financial Assumptions:") r2 = r1 + 1 val = self._df.loc[r2, c] - while not (self._is_empty(val) or val == 'Construction Duration yrs'): + while not (self._is_empty(val) or val == "Construction Duration yrs"): r2 += 1 assert r2 != self._df.shape[0], "Error finding end of fin assumptions" val = self._df.loc[r2, c] @@ -198,19 +228,21 @@ def get_fin_assump(self) -> pd.DataFrame: if self._is_empty(val): r2 -= 1 - headers = ['Financial Assumptions', 'Value'] + headers = ["Financial Assumptions", "Value"] - df_fin_assump = pd.DataFrame(self._df.loc[r1 + 1:r2, c]) - df_fin_assump['Value'] = self._df.loc[r1 + 1:r2, c + FIN_ASSUMP_COL] + df_fin_assump = pd.DataFrame(self._df.loc[r1 + 1 : r2, c]) + df_fin_assump["Value"] = self._df.loc[r1 + 1 : r2, c + FIN_ASSUMP_COL] df_fin_assump.columns = pd.Index(headers) - df_fin_assump = df_fin_assump.set_index('Financial Assumptions') + df_fin_assump = df_fin_assump.set_index("Financial Assumptions") - assert not df_fin_assump.isnull().any().any(),\ - f'Error loading financial assumptions. Found empty values: {df_fin_assump}' + assert ( + not df_fin_assump.isnull().any().any() + ), f"Error loading financial assumptions. Found empty values: {df_fin_assump}" return df_fin_assump - def get_metric_values(self, metric: str, num_tds: int, split_metrics: bool = False)\ - -> pd.DataFrame: + def get_metric_values( + self, metric: str, num_tds: int, split_metrics: bool = False + ) -> pd.DataFrame: """ Grab metric values table @@ -224,17 +256,18 @@ def get_metric_values(self, metric: str, num_tds: int, split_metrics: bool = Fal num_rows += len(self.scenarios) df_met = self._get_metric_values(metric, num_rows) - assert len(df_met) == num_tds * len(self.scenarios), \ - (f'{metric} of {self.sheet_name} ' - f'appears to be corrupt or the wrong number of tech details ({num_tds}) ' - f'was entered. split_metrics = {split_metrics}.') + assert len(df_met) == num_tds * len(self.scenarios), ( + f"{metric} of {self.sheet_name} " + f"appears to be corrupt or the wrong number of tech details ({num_tds}) " + f"was entered. split_metrics = {split_metrics}." + ) return df_met def get_tax_credits(self) -> pd.DataFrame: # HACK - 30 is arbitrary, but works - df_tc = self._get_metric_values('Tax Credit', 30) - df_tc.index.name = 'Tax Credit' + df_tc = self._get_metric_values("Tax Credit", 30) + df_tc.index.name = "Tax Credit" return df_tc def get_cff(self, cff_name: str, rows: int) -> pd.DataFrame: @@ -265,32 +298,39 @@ def _get_metric_values(self, metric, num_rows): end_col = self._next_empty_col(self._df_tech_full, r, first_col) - 1 # Extract headings - year_headings = self._df_tech_full.loc[first_row - 1, first_col + 2:end_col] + year_headings = self._df_tech_full.loc[first_row - 1, first_col + 2 : end_col] year_headings = list(year_headings.astype(int)) # Extract data df_met = self._df_tech_full.loc[first_row:end_row, first_col:end_col] - assert first_col < end_col,\ - (f'There is a formatting error for {metric} in {self.sheet_name}. ' - f'Extracted:\n{str(df_met)}') + assert first_col < end_col, ( + f"There is a formatting error for {metric} in {self.sheet_name}. " + f"Extracted:\n{str(df_met)}" + ) # Create index from tech details and cases - df_met[first_col] = df_met[first_col].astype(str) + '/' +\ - df_met[first_col + 1].astype(str) + df_met[first_col] = ( + df_met[first_col].astype(str) + "/" + df_met[first_col + 1].astype(str) + ) df_met = df_met.set_index(first_col).drop(first_col + 1, axis=1) # Clean up df_met.columns = year_headings df_met.index.name = TECH_DETAIL_SCENARIO_COL - df_met = df_met.dropna(how='all') + df_met = df_met.dropna(how="all") cols = df_met.columns - assert cols[0] == self.base_year, f'{metric}: First year should be {self.base_year}, got {cols[0]} instead' - assert cols[-1] == YEARS[-1], f'{metric}: Last year should be {YEARS[-1]}, got {cols[-1]} instead' + assert ( + cols[0] == self.base_year + ), f"{metric}: First year should be {self.base_year}, got {cols[0]} instead" + assert ( + cols[-1] == YEARS[-1] + ), f"{metric}: Last year should be {YEARS[-1]}, got {cols[-1]} instead" - assert not df_met.isnull().any().any(),\ - f'Error extracting values for {metric}. Found missing values: {df_met}' + assert ( + not df_met.isnull().any().any() + ), f"Error extracting values for {metric}. Found missing values: {df_met}" return df_met @@ -300,7 +340,7 @@ def get_meta_data(self): @returns {pd.DataFrame} """ - r, c = self._find_cell(self._df_tech_header, 'Technology Classification' ) + r, c = self._find_cell(self._df_tech_header, "Technology Classification") first_row = r + 1 first_col = c + 1 end_row = self._next_empty_row(self._df_tech_header, first_col, first_row) - 1 @@ -315,7 +355,7 @@ def get_meta_data(self): # Clean up df_meta = df_meta.reset_index(drop=True) - df_meta = df_meta.fillna('') + df_meta = df_meta.fillna("") df_meta.columns = headings return df_meta @@ -324,7 +364,7 @@ def get_meta_data(self): def _is_empty(val): if isinstance(val, str): return False - if val == '': + if val == "": return True if np.isnan(val): return True @@ -344,7 +384,7 @@ def _find_cell(df, value): assert count != 0, f'Dataframe has no instances of "{value}"' assert count <= 1, f'Dataframe has more than one instance of "{value}"' - cell = df.where(df==value).dropna(how='all').dropna(axis=1) + cell = df.where(df == value).dropna(how="all").dropna(axis=1) return cell.index[0], cell.columns[0] def _next_empty_col(self, df, row, col1): @@ -368,4 +408,4 @@ def _next_empty_row(self, df: pd.DataFrame, col: int, row1: int) -> int: row2 += 1 if row2 == len(df.loc[col]): return row2 - return row2 \ No newline at end of file + return row2 diff --git a/lcoe_calculator/macrs.py b/lcoe_calculator/macrs.py index ba0bb9f..bf9970f 100644 --- a/lcoe_calculator/macrs.py +++ b/lcoe_calculator/macrs.py @@ -8,16 +8,9 @@ MACRS Depreciation """ -MACRS_6 = [ - 0.2, - 0.32, - 0.192, - 0.1152, - 0.1152, - 0.0576 -] +MACRS_6 = [0.2, 0.32, 0.192, 0.1152, 0.1152, 0.0576] -MACRS_16=[ +MACRS_16 = [ 0.0500, 0.0950, 0.0855, @@ -33,10 +26,10 @@ 0.0591, 0.0590, 0.0591, - 0.0295 + 0.0295, ] -MACRS_21=[ +MACRS_21 = [ 0.03750, 0.07219, 0.06677, diff --git a/lcoe_calculator/process_all.py b/lcoe_calculator/process_all.py index d3d2367..243bb17 100644 --- a/lcoe_calculator/process_all.py +++ b/lcoe_calculator/process_all.py @@ -14,15 +14,26 @@ from .tech_processors import ALL_TECHS from .base_processor import TechProcessor -from .config import FINANCIAL_CASES, MARKET_FIN_CASE, CRP_CHOICES, CrpChoiceType, TAX_CREDIT_CASES +from .config import ( + FINANCIAL_CASES, + MARKET_FIN_CASE, + CRP_CHOICES, + CrpChoiceType, + TAX_CREDIT_CASES, +) + class ProcessAll: """ Extract data from ATB workbook and calculate LCOE for techs, CRPs, and financial scenarios. """ - def __init__(self, data_workbook_fname: str, - techs: List[Type[TechProcessor]]|Type[TechProcessor]): + + def __init__( + self, + data_workbook_fname: str, + techs: List[Type[TechProcessor]] | Type[TechProcessor], + ): """ @param data_workbook_fname - name of workbook @param techs - one or more techs to run @@ -36,7 +47,15 @@ def __init__(self, data_workbook_fname: str, self._techs = techs self._fname = data_workbook_fname - def _run_tech(self, Tech: TechProcessor, crp: CrpChoiceType, case : str, tcc: str, test_capex, test_lcoe): + def _run_tech( + self, + Tech: TechProcessor, + crp: CrpChoiceType, + case: str, + tcc: str, + test_capex, + test_lcoe, + ): """ Runs the specified Tech with the specified parameters @param Tech - TechProcessor to be processed @@ -62,29 +81,33 @@ def _run_tech(self, Tech: TechProcessor, crp: CrpChoiceType, case : str, tcc: st return proc def process(self, test_capex: bool = True, test_lcoe: bool = True): - """ Process all techs """ + """Process all techs""" self.data = pd.DataFrame() self.meta = pd.DataFrame() for i, Tech in enumerate(self._techs): - print(f'##### Processing {Tech.tech_name} ({i+1}/{len(self._techs)}) #####') + print(f"##### Processing {Tech.tech_name} ({i+1}/{len(self._techs)}) #####") proc = None for crp in CRP_CHOICES: # skip TechLife if 20 or 30 so we don't duplicate effort - if crp == 'TechLife' and Tech.tech_life in CRP_CHOICES: + if crp == "TechLife" and Tech.tech_life in CRP_CHOICES: continue for case in FINANCIAL_CASES: if case is MARKET_FIN_CASE and Tech.tech_name in TAX_CREDIT_CASES: tax_cases = TAX_CREDIT_CASES[Tech.tech_name] for tc in tax_cases: - proc = self._run_tech(Tech, crp, case, tc, test_capex, test_lcoe) + proc = self._run_tech( + Tech, crp, case, tc, test_capex, test_lcoe + ) else: - proc = self._run_tech(Tech, crp, case, None, test_capex, test_lcoe) + proc = self._run_tech( + Tech, crp, case, None, test_capex, test_lcoe + ) meta = proc.get_meta_data() - meta['Tech Name'] = Tech.tech_name + meta["Tech Name"] = Tech.tech_name self.meta = pd.concat([self.meta, meta]) self.data = self.data.reset_index(drop=True) @@ -92,79 +115,118 @@ def process(self, test_capex: bool = True, test_lcoe: bool = True): @property def data_flattened(self): - """ Get flat data pivoted with each year as a row """ + """Get flat data pivoted with each year as a row""" if self.data is None: - raise ValueError('Please run process() first') - - melted = pd.melt(self.data, id_vars=['Parameter', 'Case', 'TaxCreditCase', 'CRPYears', - 'Technology', 'DisplayName', 'Scenario']) + raise ValueError("Please run process() first") + + melted = pd.melt( + self.data, + id_vars=[ + "Parameter", + "Case", + "TaxCreditCase", + "CRPYears", + "Technology", + "DisplayName", + "Scenario", + ], + ) return melted def to_csv(self, fname: str): - """ Write data to CSV """ + """Write data to CSV""" if self.data is None: - raise ValueError('Please run process() first') + raise ValueError("Please run process() first") self.data.to_csv(fname) def flat_to_csv(self, fname: str): - """ Write pivoted data to CSV """ + """Write pivoted data to CSV""" if self.data is None: - raise ValueError('Please run process() first') + raise ValueError("Please run process() first") self.data_flattened.to_csv(fname) def meta_data_to_csv(self, fname: str): - """ Write meta data to CSV """ + """Write meta data to CSV""" if self.data is None: - raise ValueError('Please run process() first') + raise ValueError("Please run process() first") self.meta.to_csv(fname) tech_names = [Tech.__name__ for Tech in ALL_TECHS] + @click.command -@click.argument('data_workbook_filename', type=click.Path(exists=True)) -@click.option('-t', '--tech', type=click.Choice(tech_names), - help="Name of tech to process. Process all techs if none are specified.") -@click.option('-m', '--save-meta', 'meta_file', type=click.Path(), - help="Save meta data to CSV.") -@click.option('-f', '--save-flat', 'flat_file', type=click.Path(), - help="Save data in flat format to CSV.") -@click.option('-p', '--save-pivoted', 'pivoted_file', type=click.Path(), - help="Save data in pivoted format to CSV.") -@click.option('-c', '--clipboard', is_flag=True, default=False, - help="Copy data to system clipboard.") -def process(data_workbook_filename: str, tech: str|None, meta_file: str|None, flat_file: str|None, - pivoted_file: str|None, clipboard: bool): +@click.argument("data_workbook_filename", type=click.Path(exists=True)) +@click.option( + "-t", + "--tech", + type=click.Choice(tech_names), + help="Name of tech to process. Process all techs if none are specified.", +) +@click.option( + "-m", "--save-meta", "meta_file", type=click.Path(), help="Save meta data to CSV." +) +@click.option( + "-f", + "--save-flat", + "flat_file", + type=click.Path(), + help="Save data in flat format to CSV.", +) +@click.option( + "-p", + "--save-pivoted", + "pivoted_file", + type=click.Path(), + help="Save data in pivoted format to CSV.", +) +@click.option( + "-c", + "--clipboard", + is_flag=True, + default=False, + help="Copy data to system clipboard.", +) +def process( + data_workbook_filename: str, + tech: str | None, + meta_file: str | None, + flat_file: str | None, + pivoted_file: str | None, + clipboard: bool, +): """ CLI to process ATB data workbook and calculate metrics. """ - tech_map: Dict[str, Type[TechProcessor]] = {tech.__name__: tech for tech in ALL_TECHS} + tech_map: Dict[str, Type[TechProcessor]] = { + tech.__name__: tech for tech in ALL_TECHS + } techs = ALL_TECHS if tech is None else [tech_map[tech]] start_dt = dt.now() processor = ProcessAll(data_workbook_filename, techs) processor.process() - click.echo(f'Processing completed in {dt.now()-start_dt}.') + click.echo(f"Processing completed in {dt.now()-start_dt}.") if meta_file: - click.echo(f'Writing meta data to {meta_file}.') + click.echo(f"Writing meta data to {meta_file}.") processor.meta_data_to_csv(meta_file) if flat_file: - click.echo(f'Writing flat data to {flat_file}.') + click.echo(f"Writing flat data to {flat_file}.") processor.flat_to_csv(flat_file) if pivoted_file: - click.echo(f'Writing pivoted data to {pivoted_file}.') + click.echo(f"Writing pivoted data to {pivoted_file}.") processor.to_csv(pivoted_file) if clipboard: - click.echo('Data was copied to clipboard.') + click.echo("Data was copied to clipboard.") processor.data.to_clipboard() -if __name__ == '__main__': +if __name__ == "__main__": process() # pylint: disable=no-value-for-parameter diff --git a/lcoe_calculator/tech_extractors.py b/lcoe_calculator/tech_extractors.py index dcb1d5b..37f3dc0 100644 --- a/lcoe_calculator/tech_extractors.py +++ b/lcoe_calculator/tech_extractors.py @@ -15,13 +15,23 @@ from .config import CrpChoiceType from .extractor import Extractor + class PVBatteryExtractor(Extractor): """ Extract financial assumptions, metrics, and WACC from Excel data workbook. For the PV-plus-battery technology, with unique tax credit cases """ - def __init__(self, data_workbook_fname: str, sheet_name: str, case: str, crp: CrpChoiceType, - scenarios: List[str], base_year: int, tax_credit_case : str): + + def __init__( + self, + data_workbook_fname: str, + sheet_name: str, + case: str, + crp: CrpChoiceType, + scenarios: List[str], + base_year: int, + tax_credit_case: str, + ): """ @param data_workbook_fname - file name of data workbook @param sheet_name - name of sheet to process @@ -39,7 +49,9 @@ def __init__(self, data_workbook_fname: str, sheet_name: str, case: str, crp: Cr wb = xw.Book(data_workbook_fname) sheet = wb.sheets[sheet_name] print("Setting tax credit case", tax_credit_case) - sheet.range('Q46').value = tax_credit_case + sheet.range("Q46").value = tax_credit_case wb.save() - super().__init__(data_workbook_fname, sheet_name, case, crp, scenarios, base_year) + super().__init__( + data_workbook_fname, sheet_name, case, crp, scenarios, base_year + ) diff --git a/lcoe_calculator/tech_processors.py b/lcoe_calculator/tech_processors.py index dde94e2..d8b8c07 100644 --- a/lcoe_calculator/tech_processors.py +++ b/lcoe_calculator/tech_processors.py @@ -24,87 +24,96 @@ class OffShoreWindProc(TechProcessor): """ Abstract class, sheet name is not defined. See Fixed and Floating OSW proc """ - tech_name = 'OffShoreWind' + + tech_name = "OffShoreWind" tech_life = 30 dscr = 1.35 + class FixedOffShoreWindProc(OffShoreWindProc): - sheet_name = 'Fixed-Bottom Offshore Wind' + sheet_name = "Fixed-Bottom Offshore Wind" num_tds = 7 - default_tech_detail = 'Offshore Wind - Class 3' - wacc_name = 'Offshore Wind' + default_tech_detail = "Offshore Wind - Class 3" + wacc_name = "Offshore Wind" + class FloatingOffShoreWindProc(OffShoreWindProc): - sheet_name = 'Floating Offshore Wind' + sheet_name = "Floating Offshore Wind" num_tds = 7 base_year = 2030 - default_tech_detail = 'Offshore Wind - Class 12' - wacc_name = 'Offshore Wind' + default_tech_detail = "Offshore Wind - Class 12" + wacc_name = "Offshore Wind" + class LandBasedWindProc(TechProcessor): - tech_name = 'LandbasedWind' - sheet_name = 'Land-Based Wind' + tech_name = "LandbasedWind" + sheet_name = "Land-Based Wind" tech_life = 30 num_tds = 10 - default_tech_detail = 'Land-Based Wind - Class 4 - Technology 1' + default_tech_detail = "Land-Based Wind - Class 4 - Technology 1" dscr = 1.35 class DistributedWindProc(TechProcessor): - tech_name = 'DistributedWind' - sheet_name = 'Distributed Wind' + tech_name = "DistributedWind" + sheet_name = "Distributed Wind" tech_life = 30 num_tds = 40 - default_tech_detail = 'Midsize DW - Class 4' + default_tech_detail = "Midsize DW - Class 4" dscr = 1.35 + class UtilityPvProc(TechProcessor): - tech_name = 'UtilityPV' + tech_name = "UtilityPV" tech_life = 30 - sheet_name = 'Solar - Utility PV' + sheet_name = "Solar - Utility PV" num_tds = 10 - default_tech_detail = 'Utility PV - Class 5' + default_tech_detail = "Utility PV - Class 5" dscr = 1.25 + class CommPvProc(TechProcessor): - tech_name = 'CommPV' + tech_name = "CommPV" tech_life = 30 - sheet_name = 'Solar - PV Dist. Comm' + sheet_name = "Solar - PV Dist. Comm" num_tds = 10 - default_tech_detail = 'Commercial PV - Class 5' + default_tech_detail = "Commercial PV - Class 5" dscr = 1.25 class ResPvProc(TechProcessor): - tech_name = 'ResPV' + tech_name = "ResPV" tech_life = 30 - sheet_name = 'Solar - PV Dist. Res' + sheet_name = "Solar - PV Dist. Res" num_tds = 10 - default_tech_detail = 'Residential PV - Class 5' + default_tech_detail = "Residential PV - Class 5" dscr = 1.25 + class UtilityPvPlusBatteryProc(TechProcessor): - tech_name = 'Utility-Scale PV-Plus-Battery' + tech_name = "Utility-Scale PV-Plus-Battery" tech_life = 30 - sheet_name = 'Utility-Scale PV-Plus-Battery' + sheet_name = "Utility-Scale PV-Plus-Battery" num_tds = 10 - default_tech_detail = 'PV+Storage - Class 5' + default_tech_detail = "PV+Storage - Class 5" dscr = 1.25 - GRID_ROUNDTRIP_EFF = 0.85 # Roundtrip Efficiency (Grid charging) - CO_LOCATION_SAVINGS = 0.9228 # Reduction in OCC from co-locating the PV and battery system on the same site - BATT_PV_RATIO = 60.0 / 100.0 # Modifier for $/kW to get everything on the same basis + GRID_ROUNDTRIP_EFF = 0.85 # Roundtrip Efficiency (Grid charging) + CO_LOCATION_SAVINGS = 0.9228 # Reduction in OCC from co-locating the PV and battery system on the same site + BATT_PV_RATIO = ( + 60.0 / 100.0 + ) # Modifier for $/kW to get everything on the same basis metrics = [ - ('Net Capacity Factor (%)', 'df_ncf'), - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('PV System Cost ($/kW)', 'df_pv_cost'), - ('Battery Storage Cost ($/kW)', 'df_batt_cost'), - ('Construction Finance Factor', 'df_cff'), - ('PV-only Capacity Factor (%)','df_pvcf') + ("Net Capacity Factor (%)", "df_ncf"), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("PV System Cost ($/kW)", "df_pv_cost"), + ("Battery Storage Cost ($/kW)", "df_batt_cost"), + ("Construction Finance Factor", "df_cff"), + ("PV-only Capacity Factor (%)", "df_pvcf"), ] def __init__( @@ -113,7 +122,7 @@ def __init__( case: str = MARKET_FIN_CASE, crp: CrpChoiceType = 30, tcc: str = "PV PTC and Battery ITC", - extractor: Type[PVBatteryExtractor] = PVBatteryExtractor + extractor: Type[PVBatteryExtractor] = PVBatteryExtractor, ): # Additional data frames pulled from excel self.df_pv_cost: Optional[pd.DataFrame] = None @@ -122,38 +131,68 @@ def __init__( super().__init__(data_workbook_fname, case, crp, tcc, extractor) def _calc_lcoe(self): - batt_charge_frac = self.df_fin.loc['Fraction of Battery Energy Charged from PV (75% to 100%)', 'Value'] - grid_charge_cost = self.df_fin.loc['Average Cost of Battery Energy Charged from Grid ($/MWh)', 'Value'] + batt_charge_frac = self.df_fin.loc[ + "Fraction of Battery Energy Charged from PV (75% to 100%)", "Value" + ] + grid_charge_cost = self.df_fin.loc[ + "Average Cost of Battery Energy Charged from Grid ($/MWh)", "Value" + ] ptc = self._calc_ptc() ptc_cf_adj = self.df_pvcf / self.df_ncf - ptc_cf_adj = ptc_cf_adj.clip(upper=1.0) # account for RTE losses at 100% grid charging (might need to make equation above better) + ptc_cf_adj = ptc_cf_adj.clip( + upper=1.0 + ) # account for RTE losses at 100% grid charging (might need to make equation above better) fcr_pv = pd.concat([self.df_crf.values * self.df_pff_pv] * self.num_tds).values - fcr_batt = pd.concat([self.df_crf.values * self.df_pff_batt] * self.num_tds).values - - df_lcoe_part = (fcr_pv * self.df_cff * (self.df_pv_cost * self.CO_LOCATION_SAVINGS + self.df_gcc))\ - + (fcr_batt * self.df_cff * (self.df_batt_cost * self.CO_LOCATION_SAVINGS * self.BATT_PV_RATIO))\ - + self.df_fom - df_lcoe = (df_lcoe_part * 1000 / self.df_aep)\ - + self.df_vom\ - + (1 - batt_charge_frac) * grid_charge_cost / self.GRID_ROUNDTRIP_EFF - ptc * ptc_cf_adj + fcr_batt = pd.concat( + [self.df_crf.values * self.df_pff_batt] * self.num_tds + ).values + + df_lcoe_part = ( + ( + fcr_pv + * self.df_cff + * (self.df_pv_cost * self.CO_LOCATION_SAVINGS + self.df_gcc) + ) + + ( + fcr_batt + * self.df_cff + * (self.df_batt_cost * self.CO_LOCATION_SAVINGS * self.BATT_PV_RATIO) + ) + + self.df_fom + ) + df_lcoe = ( + (df_lcoe_part * 1000 / self.df_aep) + + self.df_vom + + (1 - batt_charge_frac) * grid_charge_cost / self.GRID_ROUNDTRIP_EFF + - ptc * ptc_cf_adj + ) return df_lcoe def _extract_data(self): - """ Pull all data from the workbook """ - crp_msg = self._requested_crp if self._requested_crp != 'TechLife' \ - else f'TechLife ({self.tech_life})' - - print(f'Loading data from {self.sheet_name}, for {self._case} and {crp_msg}') - extractor = self._ExtractorClass(self._data_workbook_fname, self.sheet_name, - self._case, self._requested_crp, self.scenarios, self.base_year, - self.tax_credit_case) - - print('\tLoading metrics') + """Pull all data from the workbook""" + crp_msg = ( + self._requested_crp + if self._requested_crp != "TechLife" + else f"TechLife ({self.tech_life})" + ) + + print(f"Loading data from {self.sheet_name}, for {self._case} and {crp_msg}") + extractor = self._ExtractorClass( + self._data_workbook_fname, + self.sheet_name, + self._case, + self._requested_crp, + self.scenarios, + self.base_year, + self.tax_credit_case, + ) + + print("\tLoading metrics") for metric, var_name in self.metrics: - if var_name == 'df_cff': + if var_name == "df_cff": # Grab DF index from another value to use in full CFF DF index = getattr(self, self.metrics[0][1]).index self.df_cff = self.load_cff(extractor, metric, index) @@ -166,30 +205,31 @@ def _extract_data(self): self.df_tc = extractor.get_tax_credits() # Pull financial assumptions from small table at top of tech sheet - print('\tLoading assumptions') + print("\tLoading assumptions") if self.has_fin_assump: self.df_fin = extractor.get_fin_assump() if self.has_wacc: - print('\tLoading WACC data') + print("\tLoading WACC data") self.df_wacc, self.df_just_wacc = extractor.get_wacc(self.wacc_name) - print('\tDone loading data') + print("\tDone loading data") return extractor def _get_tax_credit_case(self): """ - Incorporates additional code required to handle different tax credits for - different components. PV plus Battery worksheet has additional lines for + Incorporates additional code required to handle different tax credits for + different components. PV plus Battery worksheet has additional lines for battery ITC. """ - assert len(self.df_tc) > 0, \ - ('Setup df_tc with extractor.get_tax_credits() before calling this function!') + assert ( + len(self.df_tc) > 0 + ), "Setup df_tc with extractor.get_tax_credits() before calling this function!" ptc = self._calc_ptc() # Battery always takes ITC, so PV determines the case - pv_itc = self._calc_itc(itc_type=' - PV') - batt_itc = self._calc_itc(itc_type= ' - Battery') + pv_itc = self._calc_itc(itc_type=" - PV") + batt_itc = self._calc_itc(itc_type=" - Battery") # Trim the first year to eliminate pre-inflation reduction act confusion ptc = ptc[:, 1:] @@ -209,37 +249,39 @@ def _get_tax_credit_case(self): else: return "None" - def run(self): - """ Run all calculations """ + """Run all calculations""" self.df_aep = self._calc_aep() self.df_capex = self._calc_capex() self.df_cfc = self._calc_con_fin_cost() self.df_crf = self._calc_crf() - self.df_pff_pv = self._calc_pff(itc_type=' - PV') - self.df_pff_batt = self._calc_pff(itc_type=' - Battery') + self.df_pff_pv = self._calc_pff(itc_type=" - PV") + self.df_pff_batt = self._calc_pff(itc_type=" - Battery") self.df_lcoe = self._calc_lcoe() + class CspProc(TechProcessor): - tech_name = 'CSP' + tech_name = "CSP" tech_life = 30 - sheet_name = 'Solar - CSP' + sheet_name = "Solar - CSP" num_tds = 3 - default_tech_detail = 'CSP - Class 2' + default_tech_detail = "CSP - Class 2" dscr = 1.45 + class GeothermalProc(TechProcessor): - tech_name = 'Geothermal' - sheet_name = 'Geothermal' + tech_name = "Geothermal" + sheet_name = "Geothermal" tech_life = 30 num_tds = 6 - default_tech_detail = 'Geothermal - Hydro / Flash' + default_tech_detail = "Geothermal - Hydro / Flash" dscr = 1.35 @classmethod - def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, - return_short_df=False) -> pd.DataFrame: + def load_cff( + cls, extractor: Extractor, cff_name: str, index: pd.Index, return_short_df=False + ) -> pd.DataFrame: """ Special Geothermal code to load CFF and duplicate for tech details. Geothermal has 6 rows of CFF instead of the normal 3. @@ -251,9 +293,10 @@ def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, @returns - CFF data frame """ df_cff = extractor.get_cff(cff_name, len(cls.scenarios) * 2) - assert len(df_cff) == len(cls.scenarios * 2),\ - (f'Wrong number of CFF rows found. Expected {len(cls.scenarios) * 2}, ' - f'get {len(df_cff)}.') + assert len(df_cff) == len(cls.scenarios * 2), ( + f"Wrong number of CFF rows found. Expected {len(cls.scenarios) * 2}, " + f"get {len(df_cff)}." + ) if return_short_df: return df_cff @@ -269,12 +312,12 @@ def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, class HydropowerProc(TechProcessor): - tech_name = 'Hydropower' - sheet_name = 'Hydropower' + tech_name = "Hydropower" + sheet_name = "Hydropower" tech_life = 100 num_tds = 12 split_metrics = True - default_tech_detail = 'Hydropower - NPD 1' + default_tech_detail = "Hydropower - NPD 1" dscr = 1.35 def get_depreciation_schedule(self, year): @@ -283,113 +326,116 @@ def get_depreciation_schedule(self, year): else: return MACRS_6 + class PumpedStorageHydroProc(TechProcessor): - tech_name = 'Pumped Storage Hydropower' - sheet_name = 'Pumped Storage Hydropower' - wacc_name = 'Hydropower' # Use hydropower WACC values for pumped storage + tech_name = "Pumped Storage Hydropower" + sheet_name = "Pumped Storage Hydropower" + wacc_name = "Hydropower" # Use hydropower WACC values for pumped storage tech_life = 100 num_tds = 15 has_tax_credit = False has_lcoe = False flat_attrs = [ - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_capex', 'CAPEX'), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_capex", "CAPEX"), ] metrics = [ - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Construction Finance Factor', 'df_cff'), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Construction Finance Factor", "df_cff"), ] + class PumpedStorageHydroOneResProc(PumpedStorageHydroProc): - sheet_name = 'PSH One New Res' + sheet_name = "PSH One New Res" num_tds = 5 + class CoalProc(TechProcessor): - tech_name = 'Coal_FE' + tech_name = "Coal_FE" tech_life = 75 metrics = [ - ('Heat Rate (MMBtu/MWh)', 'df_hr'), - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Construction Finance Factor', 'df_cff'), + ("Heat Rate (MMBtu/MWh)", "df_hr"), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Construction Finance Factor", "df_cff"), ] flat_attrs = [ - ('df_hr', 'Heat Rate'), - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_capex', 'CAPEX'), + ("df_hr", "Heat Rate"), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_capex", "CAPEX"), ] - sheet_name = 'Coal_FE' + sheet_name = "Coal_FE" num_tds = 5 has_tax_credit = False has_lcoe = False - default_tech_detail = 'Coal-95%-CCS' + default_tech_detail = "Coal-95%-CCS" dscr = 1.45 _depreciation_schedule = MACRS_21 class NaturalGasProc(TechProcessor): - tech_name = 'NaturalGas_FE' + tech_name = "NaturalGas_FE" tech_life = 55 metrics = [ - ('Heat Rate (MMBtu/MWh)', 'df_hr'), - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Construction Finance Factor', 'df_cff'), + ("Heat Rate (MMBtu/MWh)", "df_hr"), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Construction Finance Factor", "df_cff"), ] flat_attrs = [ - ('df_hr', 'Heat Rate'), - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_capex', 'CAPEX'), + ("df_hr", "Heat Rate"), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_capex", "CAPEX"), ] - sheet_name = 'Natural Gas_FE' + sheet_name = "Natural Gas_FE" num_tds = 10 has_tax_credit = False has_lcoe = False - default_tech_detail = 'NG F-Frame CC 95% CCS' + default_tech_detail = "NG F-Frame CC 95% CCS" dscr = 1.45 _depreciation_schedule = MACRS_21 class NaturalGasFuelCellProc(NaturalGasProc): - sheet_name = 'Natural Gas Fuel Cell_FE' + sheet_name = "Natural Gas Fuel Cell_FE" num_tds = 2 has_wacc = False has_lcoe = False has_fin_assump = False - default_tech_detail = 'NG Fuel Cell Max CCS' + default_tech_detail = "NG Fuel Cell Max CCS" - scenarios = ['Moderate', 'Advanced'] + scenarios = ["Moderate", "Advanced"] base_year = 2035 class CoalRetrofitProc(TechProcessor): - tech_name = 'Coal_Retrofits' + tech_name = "Coal_Retrofits" tech_life = 75 has_wacc = False @@ -398,30 +444,30 @@ class CoalRetrofitProc(TechProcessor): has_fin_assump = False metrics = [ - ('Heat Rate (MMBtu/MWh)', 'df_hr'), - ('Additional Overnight Capital Cost ($/kW)', 'df_occ'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Heat Rate Penalty (Δ% from pre-retrofit)' , 'df_hrp'), - ('Net Output Penalty (Δ% from pre-retrofit)' , 'df_nop') + ("Heat Rate (MMBtu/MWh)", "df_hr"), + ("Additional Overnight Capital Cost ($/kW)", "df_occ"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Heat Rate Penalty (Δ% from pre-retrofit)", "df_hrp"), + ("Net Output Penalty (Δ% from pre-retrofit)", "df_nop"), ] flat_attrs = [ - ('df_hr', 'Heat Rate'), - ('df_occ', 'Additional OCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_hrp', 'Heat Rate Penalty'), - ('df_nop', 'Net Output Penalty') + ("df_hr", "Heat Rate"), + ("df_occ", "Additional OCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_hrp", "Heat Rate Penalty"), + ("df_nop", "Net Output Penalty"), ] - sheet_name = 'Coal_Retrofits' + sheet_name = "Coal_Retrofits" num_tds = 2 has_tax_credit = False class NaturalGasRetrofitProc(TechProcessor): - tech_name = 'NaturalGas_Retrofits' + tech_name = "NaturalGas_Retrofits" tech_life = 55 has_wacc = False @@ -430,64 +476,65 @@ class NaturalGasRetrofitProc(TechProcessor): has_fin_assump = False metrics = [ - ('Heat Rate (MMBtu/MWh)', 'df_hr'), - ('Additional Overnight Capital Cost ($/kW)', 'df_occ'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Heat Rate Penalty (Δ% from pre-retrofit)' , 'df_hrp'), - ('Net Output Penalty (Δ% from pre-retrofit)' , 'df_nop') + ("Heat Rate (MMBtu/MWh)", "df_hr"), + ("Additional Overnight Capital Cost ($/kW)", "df_occ"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Heat Rate Penalty (Δ% from pre-retrofit)", "df_hrp"), + ("Net Output Penalty (Δ% from pre-retrofit)", "df_nop"), ] flat_attrs = [ - ('df_hr', 'Heat Rate'), - ('df_occ', 'Additional OCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_hrp', 'Heat Rate Penalty'), - ('df_nop', 'Net Output Penalty') + ("df_hr", "Heat Rate"), + ("df_occ", "Additional OCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_hrp", "Heat Rate Penalty"), + ("df_nop", "Net Output Penalty"), ] - sheet_name = 'Natural Gas_Retrofits' + sheet_name = "Natural Gas_Retrofits" num_tds = 4 has_tax_credit = False class NuclearProc(TechProcessor): - tech_name = 'Nuclear' + tech_name = "Nuclear" tech_life = 60 - sheet_name = 'Nuclear' + sheet_name = "Nuclear" num_tds = 2 - default_tech_detail = 'Nuclear - Large' + default_tech_detail = "Nuclear - Large" dscr = 1.45 base_year = 2030 metrics = [ - ('Heat Rate (MMBtu/MWh)', 'df_hr'), - ('Net Capacity Factor (%)', 'df_ncf'), - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Fuel Costs ($/MMBtu)', 'df_fuel_costs_mmbtu'), - ('Construction Finance Factor', 'df_cff'), + ("Heat Rate (MMBtu/MWh)", "df_hr"), + ("Net Capacity Factor (%)", "df_ncf"), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Fuel Costs ($/MMBtu)", "df_fuel_costs_mmbtu"), + ("Construction Finance Factor", "df_cff"), ] flat_attrs = [ - ('df_ncf', 'CF'), - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_lcoe', 'LCOE'), - ('df_capex', 'CAPEX'), - ('df_fuel_costs_mwh', 'Fuel'), - ('df_hr', 'Heat Rate'), + ("df_ncf", "CF"), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_lcoe", "LCOE"), + ("df_capex", "CAPEX"), + ("df_fuel_costs_mwh", "Fuel"), + ("df_hr", "Heat Rate"), ] @classmethod - def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, - return_short_df=False) -> pd.DataFrame: + def load_cff( + cls, extractor: Extractor, cff_name: str, index: pd.Index, return_short_df=False + ) -> pd.DataFrame: """ Load CFF data from workbook. Nuclear has a unique CFF for each tech detail, so this function removes the tech detail duplication code from BaseProcessor. @@ -499,7 +546,7 @@ def load_cff(cls, extractor: Extractor, cff_name: str, index: pd.Index, return df_cff def _calc_lcoe(self): - """ Include fuel costs in LCOE """ + """Include fuel costs in LCOE""" # pylint: disable=no-member,attribute-defined-outside-init self.df_fuel_costs_mwh = self.df_hr * self.df_fuel_costs_mmbtu df_lcoe = super()._calc_lcoe() + self.df_fuel_costs_mwh @@ -513,39 +560,39 @@ def get_depreciation_schedule(self, year): class BiopowerProc(TechProcessor): - tech_name = 'Biopower' + tech_name = "Biopower" tech_life = 45 - sheet_name = 'Biopower' + sheet_name = "Biopower" num_tds = 1 - default_tech_detail = 'Biopower - Dedicated' + default_tech_detail = "Biopower - Dedicated" dscr = 1.45 metrics = [ - ('Heat Rate (MMBtu/MWh)', 'df_hr'), - ('Net Capacity Factor (%)', 'df_ncf'), - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Fuel Costs ($/MMBtu)', 'df_fuel_costs_mmbtu'), - ('Construction Finance Factor', 'df_cff'), + ("Heat Rate (MMBtu/MWh)", "df_hr"), + ("Net Capacity Factor (%)", "df_ncf"), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Fuel Costs ($/MMBtu)", "df_fuel_costs_mmbtu"), + ("Construction Finance Factor", "df_cff"), ] flat_attrs = [ - ('df_ncf', 'CF'), - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_lcoe', 'LCOE'), - ('df_capex', 'CAPEX'), - ('df_fuel_costs_mwh', 'Fuel'), - ('df_hr', 'Heat Rate'), + ("df_ncf", "CF"), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_lcoe", "LCOE"), + ("df_capex", "CAPEX"), + ("df_fuel_costs_mwh", "Fuel"), + ("df_hr", "Heat Rate"), ] def _calc_lcoe(self): - """ Include fuel costs in LCOE """ + """Include fuel costs in LCOE""" # pylint: disable=no-member,attribute-defined-outside-init self.df_fuel_costs_mwh = self.df_hr * self.df_fuel_costs_mmbtu df_lcoe = super()._calc_lcoe() + self.df_fuel_costs_mwh @@ -557,6 +604,7 @@ class AbstractBatteryProc(TechProcessor): """ Abstract tech processor for batteries w/o LCOE or CAPEX. """ + has_wacc = False has_lcoe = False @@ -564,52 +612,68 @@ class AbstractBatteryProc(TechProcessor): has_tax_credit = False metrics = [ - ('Overnight Capital Cost ($/kW)', 'df_occ'), - ('Grid Connection Costs (GCC) ($/kW)', 'df_gcc'), - ('Fixed Operation and Maintenance Expenses ($/kW-yr)', 'df_fom'), - ('Variable Operation and Maintenance Expenses ($/MWh)', 'df_vom'), - ('Construction Finance Factor', 'df_cff'), + ("Overnight Capital Cost ($/kW)", "df_occ"), + ("Grid Connection Costs (GCC) ($/kW)", "df_gcc"), + ("Fixed Operation and Maintenance Expenses ($/kW-yr)", "df_fom"), + ("Variable Operation and Maintenance Expenses ($/MWh)", "df_vom"), + ("Construction Finance Factor", "df_cff"), ] flat_attrs = [ - ('df_occ', 'OCC'), - ('df_gcc', 'GCC'), - ('df_fom', 'Fixed O&M'), - ('df_vom', 'Variable O&M'), - ('df_cfc', 'CFC'), - ('df_capex', 'CAPEX'), + ("df_occ", "OCC"), + ("df_gcc", "GCC"), + ("df_fom", "Fixed O&M"), + ("df_vom", "Variable O&M"), + ("df_cfc", "CFC"), + ("df_capex", "CAPEX"), ] class UtilityBatteryProc(AbstractBatteryProc): - tech_name = 'Utility-Scale Battery Storage' + tech_name = "Utility-Scale Battery Storage" tech_life = 30 - sheet_name = 'Utility-Scale Battery Storage' + sheet_name = "Utility-Scale Battery Storage" num_tds = 5 class CommBatteryProc(AbstractBatteryProc): - tech_name = 'Commercial Battery Storage' + tech_name = "Commercial Battery Storage" tech_life = 30 - sheet_name = 'Commercial Battery Storage' + sheet_name = "Commercial Battery Storage" num_tds = 5 class ResBatteryProc(AbstractBatteryProc): - tech_name = 'Residential Battery Storage' + tech_name = "Residential Battery Storage" tech_life = 30 - sheet_name = 'Residential Battery Storage' + sheet_name = "Residential Battery Storage" num_tds = 2 -ALL_TECHS: List[Type[TechProcessor]]= [ - FixedOffShoreWindProc, FloatingOffShoreWindProc, LandBasedWindProc, DistributedWindProc, - UtilityPvProc, CommPvProc, ResPvProc, UtilityPvPlusBatteryProc, - CspProc, GeothermalProc, HydropowerProc, PumpedStorageHydroProc, +ALL_TECHS: List[Type[TechProcessor]] = [ + FixedOffShoreWindProc, + FloatingOffShoreWindProc, + LandBasedWindProc, + DistributedWindProc, + UtilityPvProc, + CommPvProc, + ResPvProc, + UtilityPvPlusBatteryProc, + CspProc, + GeothermalProc, + HydropowerProc, + PumpedStorageHydroProc, PumpedStorageHydroOneResProc, - CoalProc, NaturalGasProc, NuclearProc, BiopowerProc, - UtilityBatteryProc, CommBatteryProc, ResBatteryProc, - CoalRetrofitProc, NaturalGasRetrofitProc, NaturalGasFuelCellProc + CoalProc, + NaturalGasProc, + NuclearProc, + BiopowerProc, + UtilityBatteryProc, + CommBatteryProc, + ResBatteryProc, + CoalRetrofitProc, + NaturalGasRetrofitProc, + NaturalGasFuelCellProc, ] diff --git a/tests/data_finder.py b/tests/data_finder.py index 9c7d5fe..1c2f445 100644 --- a/tests/data_finder.py +++ b/tests/data_finder.py @@ -20,14 +20,14 @@ from lcoe_calculator.base_processor import TechProcessor from lcoe_calculator.config import LCOE_SS_NAME, CAPEX_SS_NAME, CrpChoiceType -DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data') +DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") # These metrics do not have real headers in the data workbook. Create fake ones so data can be # stored for testing purposes. -FIN_ASSUMP_FAKE_SS_NAME = 'financial assumptions' -WACC_FAKE_SS_NAME = 'wacc' -JUST_WACC_FAKE_SS_NAME = 'just wacc' -TAX_CREDIT_FAKE_SS_NAME = 'tax credit' +FIN_ASSUMP_FAKE_SS_NAME = "financial assumptions" +WACC_FAKE_SS_NAME = "wacc" +JUST_WACC_FAKE_SS_NAME = "just wacc" +TAX_CREDIT_FAKE_SS_NAME = "tax credit" class DataFinder: @@ -35,6 +35,7 @@ class DataFinder: Get path and file names for saving tech metric values to CSV. Note that set_tech() must be used before first use and before being used for a new tech. """ + _tech: Optional[Type[TechProcessor]] = None @classmethod @@ -56,40 +57,43 @@ def get_data_filename(cls, metric: str, case: str, crp: CrpChoiceType): @param crp - name of desired CRP @returns path to CSV file for metric in testing data dir """ - assert cls._tech is not None, 'The TechProcessor must be set first with set_tech().' + assert ( + cls._tech is not None + ), "The TechProcessor must be set first with set_tech()." # Create a lookup table between fancy long names in the workbook and names to use for the # data files. This table partially borrows from the metrics list. metric_lookup = list(cls._tech.metrics) metric_lookup += [ - (LCOE_SS_NAME, 'df_lcoe'), - (CAPEX_SS_NAME, 'df_capex'), - (FIN_ASSUMP_FAKE_SS_NAME, 'df_fin_assump'), - (WACC_FAKE_SS_NAME, 'df_wacc'), - (JUST_WACC_FAKE_SS_NAME, 'df_just_wacc'), - (TAX_CREDIT_FAKE_SS_NAME, 'df_tc'), + (LCOE_SS_NAME, "df_lcoe"), + (CAPEX_SS_NAME, "df_capex"), + (FIN_ASSUMP_FAKE_SS_NAME, "df_fin_assump"), + (WACC_FAKE_SS_NAME, "df_wacc"), + (JUST_WACC_FAKE_SS_NAME, "df_just_wacc"), + (TAX_CREDIT_FAKE_SS_NAME, "df_tc"), ] - assert metric in [m[0] for m in metric_lookup],\ - f'metric {metric} is not known for sheet {cls._tech.sheet_name}' + assert metric in [ + m[0] for m in metric_lookup + ], f"metric {metric} is not known for sheet {cls._tech.sheet_name}" df_name = [m[1] for m in metric_lookup if m[0] == metric][0] # Files in ./data/{tech} - clean_sheet_name = str(cls._tech.sheet_name).replace(' ', '_') + clean_sheet_name = str(cls._tech.sheet_name).replace(" ", "_") tech_dir = os.path.join(DATA_DIR, clean_sheet_name) if not os.path.exists(tech_dir): os.makedirs(tech_dir) - if df_name in ['df_ncf']: - return os.path.join(tech_dir, f'{df_name}.csv') + if df_name in ["df_ncf"]: + return os.path.join(tech_dir, f"{df_name}.csv") # Files in ./data/{tech}/{case} case_dir = os.path.join(tech_dir, case) if not os.path.exists(case_dir): os.makedirs(case_dir) - if df_name in ['df_cff', 'df_wacc', 'df_just_wacc']: - return os.path.join(case_dir, f'{df_name}.csv') + if df_name in ["df_cff", "df_wacc", "df_just_wacc"]: + return os.path.join(case_dir, f"{df_name}.csv") # Files in ./data/{tech}/{case}/{crp} crp_dir = os.path.join(case_dir, str(crp)) if not os.path.exists(crp_dir): os.makedirs(crp_dir) - return os.path.join(crp_dir, f'{df_name}.csv') + return os.path.join(crp_dir, f"{df_name}.csv") diff --git a/tests/extract_test_data.py b/tests/extract_test_data.py index bb06cfd..bf0b38d 100644 --- a/tests/extract_test_data.py +++ b/tests/extract_test_data.py @@ -13,14 +13,27 @@ from lcoe_calculator.base_processor import TechProcessor from lcoe_calculator.tech_processors import ALL_TECHS from lcoe_calculator.extractor import Extractor -from lcoe_calculator.config import FINANCIAL_CASES, LCOE_SS_NAME, CAPEX_SS_NAME, CFF_SS_NAME,\ - CRP_CHOICES, CrpChoiceType -from .data_finder import DataFinder, FIN_ASSUMP_FAKE_SS_NAME, WACC_FAKE_SS_NAME,\ - JUST_WACC_FAKE_SS_NAME, TAX_CREDIT_FAKE_SS_NAME +from lcoe_calculator.config import ( + FINANCIAL_CASES, + LCOE_SS_NAME, + CAPEX_SS_NAME, + CFF_SS_NAME, + CRP_CHOICES, + CrpChoiceType, +) +from .data_finder import ( + DataFinder, + FIN_ASSUMP_FAKE_SS_NAME, + WACC_FAKE_SS_NAME, + JUST_WACC_FAKE_SS_NAME, + TAX_CREDIT_FAKE_SS_NAME, +) + # Use extractor to pull values from data workbook and save as CSV -def extract_data_for_crp_case(data_workbook_fname: str, tech: Type[TechProcessor], case: str, - crp: CrpChoiceType): +def extract_data_for_crp_case( + data_workbook_fname: str, tech: Type[TechProcessor], case: str, crp: CrpChoiceType +): """ Extract data from ATB data workbook for a tech and save as CSV. @@ -29,19 +42,25 @@ def extract_data_for_crp_case(data_workbook_fname: str, tech: Type[TechProcessor @param case - name of desired financial case @param crp - name of desired CRP """ - extractor = Extractor(data_workbook_fname, str(tech.sheet_name), case, crp, - tech.scenarios, base_year=tech.base_year) + extractor = Extractor( + data_workbook_fname, + str(tech.sheet_name), + case, + crp, + tech.scenarios, + base_year=tech.base_year, + ) metrics = list(tech.metrics) if tech.has_lcoe: - metrics.append((LCOE_SS_NAME, '')) + metrics.append((LCOE_SS_NAME, "")) if tech.has_capex: - metrics.append((CAPEX_SS_NAME, '')) + metrics.append((CAPEX_SS_NAME, "")) extract_cff = False - for metric, _ in metrics : + for metric, _ in metrics: if metric == CFF_SS_NAME: extract_cff = True continue @@ -76,15 +95,18 @@ def extract_data_for_crp_case(data_workbook_fname: str, tech: Type[TechProcessor tech_names = [tech.__name__ for tech in ALL_TECHS] + @click.command -@click.argument('filename', type=click.Path(exists=True)) -@click.option('-t', '--tech', type=click.Choice(tech_names)) -def extract(filename: str, tech: str|None): +@click.argument("filename", type=click.Path(exists=True)) +@click.option("-t", "--tech", type=click.Choice(tech_names)) +def extract(filename: str, tech: str | None): """ Extract test data for one or more techs for all CRPs and financial cases. Data will be extracted from the Excel ATB data workbook FILENAME and saved as CSV for testing. """ - tech_map: Dict[str, Type[TechProcessor]] = {tech.__name__: tech for tech in ALL_TECHS} + tech_map: Dict[str, Type[TechProcessor]] = { + tech.__name__: tech for tech in ALL_TECHS + } if tech is None: techs = ALL_TECHS @@ -92,16 +114,16 @@ def extract(filename: str, tech: str|None): techs = [tech_map[tech]] for Tech in techs: - print(f'Extracting values for {Tech.sheet_name}') + print(f"Extracting values for {Tech.sheet_name}") DataFinder.set_tech(Tech) for case in FINANCIAL_CASES: for crp in CRP_CHOICES: - print(f'\tcrp={crp}, case={case}') + print(f"\tcrp={crp}, case={case}") extract_data_for_crp_case(filename, Tech, case, crp) - print('Done') + print("Done") -if __name__ == '__main__': +if __name__ == "__main__": extract() # pylint: disable=no-value-for-parameter diff --git a/tests/mock_extractor.py b/tests/mock_extractor.py index 5b7f63f..c41ae89 100644 --- a/tests/mock_extractor.py +++ b/tests/mock_extractor.py @@ -11,8 +11,13 @@ import pandas as pd from lcoe_calculator.abstract_extractor import AbstractExtractor from lcoe_calculator.config import CrpChoiceType -from .data_finder import DataFinder, TAX_CREDIT_FAKE_SS_NAME, WACC_FAKE_SS_NAME,\ - JUST_WACC_FAKE_SS_NAME, FIN_ASSUMP_FAKE_SS_NAME +from .data_finder import ( + DataFinder, + TAX_CREDIT_FAKE_SS_NAME, + WACC_FAKE_SS_NAME, + JUST_WACC_FAKE_SS_NAME, + FIN_ASSUMP_FAKE_SS_NAME, +) class MockExtractor(AbstractExtractor): @@ -24,8 +29,16 @@ class MockExtractor(AbstractExtractor): before MockExtractor is used to to load data from the data directory. """ - def __init__(self, _: str, __: str, case: str, crp: CrpChoiceType, ___: List[int], - ____: int, _____: Optional[str] = None): + def __init__( + self, + _: str, + __: str, + case: str, + crp: CrpChoiceType, + ___: List[int], + ____: int, + _____: Optional[str] = None, + ): """ @param data_workbook_fname - IGNORED @param sheet_name - IGNORED @@ -38,7 +51,7 @@ def __init__(self, _: str, __: str, case: str, crp: CrpChoiceType, ___: List[int self._case = case self._requested_crp = crp - def get_metric_values(self, metric: str, _:int, __=False) -> pd.DataFrame: + def get_metric_values(self, metric: str, _: int, __=False) -> pd.DataFrame: """ Grab metric values table @@ -52,8 +65,10 @@ def get_metric_values(self, metric: str, _:int, __=False) -> pd.DataFrame: return df def get_tax_credits(self) -> pd.DataFrame: - """ Get tax credit """ - fname = DataFinder.get_data_filename(TAX_CREDIT_FAKE_SS_NAME, self._case, self._requested_crp) + """Get tax credit""" + fname = DataFinder.get_data_filename( + TAX_CREDIT_FAKE_SS_NAME, self._case, self._requested_crp + ) df = self.read_csv(fname) return df @@ -74,7 +89,9 @@ def get_fin_assump(self) -> pd.DataFrame: Dynamically search for financial assumptions in small table at top of tech sheet and return as data frame """ - fname = DataFinder.get_data_filename(FIN_ASSUMP_FAKE_SS_NAME, self._case, self._requested_crp) + fname = DataFinder.get_data_filename( + FIN_ASSUMP_FAKE_SS_NAME, self._case, self._requested_crp + ) df = pd.read_csv(fname, index_col=0) return df @@ -87,9 +104,13 @@ def get_wacc(self, _=None) -> Tuple[pd.DataFrame, pd.DataFrame]: @returns {pd.DataFrame} df_just_wacc - last six rows of wacc sheet, 'WACC Nominal - {scenario}' and 'WACC Real - {scenario}' """ - fname = DataFinder.get_data_filename(WACC_FAKE_SS_NAME, self._case, self._requested_crp) + fname = DataFinder.get_data_filename( + WACC_FAKE_SS_NAME, self._case, self._requested_crp + ) df_wacc = self.read_csv(fname) - fname = DataFinder.get_data_filename(JUST_WACC_FAKE_SS_NAME, self._case, self._requested_crp) + fname = DataFinder.get_data_filename( + JUST_WACC_FAKE_SS_NAME, self._case, self._requested_crp + ) df_just_wacc = self.read_csv(fname) return (df_wacc, df_just_wacc) diff --git a/tests/test_debt_fraction_calculator.py b/tests/test_debt_fraction_calculator.py index 03b0847..b548274 100644 --- a/tests/test_debt_fraction_calculator.py +++ b/tests/test_debt_fraction_calculator.py @@ -14,93 +14,96 @@ def test_no_tax_credits(): - """ 2023 ATB utility PV, 2030, R&D case """ + """2023 ATB utility PV, 2030, R&D case""" input_vals = { - "CF" : 0.29485, - "OCC" : 1043.0, - "CFC" : 38.0, - "Fixed O&M" : 18.0, - "Variable O&M" : 0.0, - "DSCR" : 1.3, - "Rate of Return on Equity Nominal" : 0.088, - "Tax Rate (Federal and State)" : 0.257, - "Inflation Rate" : 0.025, - "Interest Rate Nominal" : 0.07, - "Calculated Rate of Return on Equity Real" : 0.061, - "ITC" : 0, - "PTC" : 0, - "MACRS" : MACRS_6 + "CF": 0.29485, + "OCC": 1043.0, + "CFC": 38.0, + "Fixed O&M": 18.0, + "Variable O&M": 0.0, + "DSCR": 1.3, + "Rate of Return on Equity Nominal": 0.088, + "Tax Rate (Federal and State)": 0.257, + "Inflation Rate": 0.025, + "Interest Rate Nominal": 0.07, + "Calculated Rate of Return on Equity Real": 0.061, + "ITC": 0, + "PTC": 0, + "MACRS": MACRS_6, } debt_frac = calculate_debt_fraction(input_vals) assert debt_frac == pytest.approx(73.8, 0.1) + def test_ptc(): - """ 2023 ATB utility PV, 2030, Markets case """ + """2023 ATB utility PV, 2030, Markets case""" input_vals = { - "CF" : 0.29485, - "OCC" : 1043.0, - "CFC" : 38.0, - "Fixed O&M" : 18.0, - "Variable O&M" : 0.0, - "DSCR" : 1.3, - "Rate of Return on Equity Nominal" : 0.088, - "Tax Rate (Federal and State)" : 0.257, - "Inflation Rate" : 0.025, - "Interest Rate Nominal" : 0.07, - "Calculated Rate of Return on Equity Real" : 0.061, - "ITC" : 0, - "PTC" : 25.46, - "MACRS" : MACRS_6 + "CF": 0.29485, + "OCC": 1043.0, + "CFC": 38.0, + "Fixed O&M": 18.0, + "Variable O&M": 0.0, + "DSCR": 1.3, + "Rate of Return on Equity Nominal": 0.088, + "Tax Rate (Federal and State)": 0.257, + "Inflation Rate": 0.025, + "Interest Rate Nominal": 0.07, + "Calculated Rate of Return on Equity Real": 0.061, + "ITC": 0, + "PTC": 25.46, + "MACRS": MACRS_6, } debt_frac = calculate_debt_fraction(input_vals) assert debt_frac == pytest.approx(45.5, 0.1) + def test_itc(): - """ 2023 ATB utility PV, 2030, Markets case """ + """2023 ATB utility PV, 2030, Markets case""" input_vals = { - "CF" : 0.29485, - "OCC" : 1043.0, - "CFC" : 38.0, - "Fixed O&M" : 18.0, - "Variable O&M" : 0.0, - "DSCR" : 1.3, - "Rate of Return on Equity Nominal" : 0.088, - "Tax Rate (Federal and State)" : 0.257, - "Inflation Rate" : 0.025, - "Interest Rate Nominal" : 0.07, - "Calculated Rate of Return on Equity Real" : 0.061, - "ITC" : 0.3, - "PTC" : 0, - "MACRS" : MACRS_6 + "CF": 0.29485, + "OCC": 1043.0, + "CFC": 38.0, + "Fixed O&M": 18.0, + "Variable O&M": 0.0, + "DSCR": 1.3, + "Rate of Return on Equity Nominal": 0.088, + "Tax Rate (Federal and State)": 0.257, + "Inflation Rate": 0.025, + "Interest Rate Nominal": 0.07, + "Calculated Rate of Return on Equity Real": 0.061, + "ITC": 0.3, + "PTC": 0, + "MACRS": MACRS_6, } debt_frac = calculate_debt_fraction(input_vals) assert debt_frac == pytest.approx(51.8, 0.1) + def test_heat_rate(): - """ Nuclear, 2030 """ + """Nuclear, 2030""" input_vals = { - "CF" : 0.93, - "OCC" : 6115.0, - "CFC" : 1615.0, - "Fixed O&M" : 152.0, - "Variable O&M" : 2.0, - "DSCR" : 1.45, - "Rate of Return on Equity Nominal" : 0.11, - "Tax Rate (Federal and State)" : 0.257, - "Inflation Rate" : 0.025, - "Interest Rate Nominal" : 0.08, - "Calculated Rate of Return on Equity Real" : 0.083, - "ITC" : 0.3, - "PTC" : 0, - "MACRS" : MACRS_6, - "Fuel" : 7.0, - "Heat Rate" : 10.45 + "CF": 0.93, + "OCC": 6115.0, + "CFC": 1615.0, + "Fixed O&M": 152.0, + "Variable O&M": 2.0, + "DSCR": 1.45, + "Rate of Return on Equity Nominal": 0.11, + "Tax Rate (Federal and State)": 0.257, + "Inflation Rate": 0.025, + "Interest Rate Nominal": 0.08, + "Calculated Rate of Return on Equity Real": 0.083, + "ITC": 0.3, + "PTC": 0, + "MACRS": MACRS_6, + "Fuel": 7.0, + "Heat Rate": 10.45, } debt_frac = calculate_debt_fraction(input_vals) diff --git a/tests/test_lcoe_calculator.py b/tests/test_lcoe_calculator.py index a8c62c2..1d0d78c 100644 --- a/tests/test_lcoe_calculator.py +++ b/tests/test_lcoe_calculator.py @@ -17,18 +17,23 @@ from .mock_extractor import MockExtractor from .data_finder import DataFinder + def test_lcoe_and_capex_calculations(): """ Test LCOE and CAPEX calculations using stored data """ for Tech in ALL_TECHS: - print(f'----------- Testing {Tech.sheet_name} -----------') + print(f"----------- Testing {Tech.sheet_name} -----------") for case in FINANCIAL_CASES: for crp in CRP_CHOICES: DataFinder.set_tech(Tech) - proc: TechProcessor = Tech('fake_path_to_data_workbook.xlsx', case=case, crp=crp, - extractor=MockExtractor) + proc: TechProcessor = Tech( + "fake_path_to_data_workbook.xlsx", + case=case, + crp=crp, + extractor=MockExtractor, + ) proc.run() # Check all metrics have been loaded @@ -49,15 +54,19 @@ def test_lcoe_and_capex_calculations(): proc.test_capex() assert not proc.df_capex.isnull().any().any() assert not proc.ss_capex.isnull().any().any() - assert np.allclose(np.array(proc.df_capex, dtype=float), - np.array(proc.ss_capex, dtype=float)) + assert np.allclose( + np.array(proc.df_capex, dtype=float), + np.array(proc.ss_capex, dtype=float), + ) if proc.has_lcoe: proc.test_lcoe() assert not proc.df_lcoe.isnull().any().any() assert not proc.ss_lcoe.isnull().any().any() - assert np.allclose(np.array(proc.df_lcoe, dtype=float), - np.array(proc.ss_lcoe, dtype=float)) + assert np.allclose( + np.array(proc.df_lcoe, dtype=float), + np.array(proc.ss_lcoe, dtype=float), + ) -if __name__ == '__main__': +if __name__ == "__main__": test_lcoe_and_capex_calculations()