Source code for postproc_setup
from pathlib import Path
import pandas as pd
[docs]class SetPaths:
"""
The SetPaths class sets the paths and run_id info used by the tool.
"""
def __init__(self):
# set paths
self.path_code = Path(__file__).parent
self.path_tool = self.path_code.parent
self.path_project = self.path_tool.parent
self.path_inputs = self.path_tool / 'inputs'
self.path_tool_runs = self.path_tool / 'runs'
self.path_tool_runs.mkdir(exist_ok=True)
# create generator of files in path_code
self.files_in_path_code = (entry for entry in self.path_code.iterdir() if entry.is_file())
# create generator of files in the tool's inputs folder
self.files_in_tool_inputs = (entry for entry in self.path_inputs.iterdir() if entry.is_file())
[docs] def set_results_folders(self, run_folder_identifier):
"""
Parameters:
run_folder_identifier: String including a timestamp set in-code and, if selected, identifier info supplied by the user via console prompt.
Return:
Path to results associated with the given model run.
"""
path_tool_runs_runid = self.path_tool / 'runs' / f'{run_folder_identifier}'
path_tool_runs_runid.mkdir(exist_ok=False)
path_tool_runs_runid_outputs = path_tool_runs_runid / 'postproc_outputs'
path_tool_runs_runid_outputs.mkdir(exist_ok=False)
path_tool_runs_runid_code = path_tool_runs_runid / 'code'
path_tool_runs_runid_code.mkdir(exist_ok=False)
path_tool_runs_runid_inputs = path_tool_runs_runid / 'inputs'
path_tool_runs_runid_inputs.mkdir(exist_ok=False)
return path_tool_runs_runid, path_tool_runs_runid_inputs, path_tool_runs_runid_outputs, path_tool_runs_runid_code
[docs] @staticmethod
def get_run_identifiers(time_of_postproc_run):
"""
Parameters:
time_of_postproc_run: String set in-code denoting the timestamp at the start of the given tool run.
Return:
run_id set by the user, if desired, in response to a console prompt \n
run_folder_identifier consisting of the passed timestamp and, if set by the user, the run_id \n
filename_id to be included in all files generated by this tool \n
run_details to be included in the summary_log for the given tool run \n
"""
run_id = input('Provide a run identifier for the output folder name (press return to use the default name)\n')
run_folder_identifier = f'{time_of_postproc_run}_{run_id}' if run_id != '' else f'{time_of_postproc_run}'
filename_id = f'{run_id}' if run_id != '' else f'{time_of_postproc_run}'
run_details = input('Provide some summary details for the given run if desired. Press <ENTER> if not.\n')
return run_id, run_folder_identifier, filename_id, run_details
[docs]class RuntimeSettings:
"""
The RuntimeSettings class reads the runtime_settings.csv input file and parses the information to establish what CCEMS reports to post-process.
"""
def __init__(self):
set_paths = SetPaths()
self.runtime_dict = pd.read_csv(set_paths.path_inputs / 'runtime_settings.csv', index_col=0).to_dict('index')
self.run_compliance_report = True if self.runtime_dict['run_compliance_report']['value'] == 1 else False
self.run_effects_summary_report = True if self.runtime_dict['run_effects_summary_report']['value'] == 1 else False
self.run_costs_summary_report = True if self.runtime_dict['run_costs_summary_report']['value'] == 1 else False
self.run_effects_report = True if self.runtime_dict['run_effects_report']['value'] == 1 else False
self.run_costs_report = True if self.runtime_dict['run_costs_report']['value'] == 1 else False
self.run_tech_utilization_report = True if self.runtime_dict['run_tech_utilization_report']['value'] == 1 else False
self.run_vehicles_report = True if self.runtime_dict['run_vehicles_report']['value'] == 1 else False
self.run_copy_paster = True if self.runtime_dict['run_copy_paster']['value'] == 1 else False
[docs]class SetInputs:
"""
The SetInputs class reads the general_inputs.csv input file and parses the information needed for the given tool run. This class also reads the cost factors inputs files and creates the
dictionaries needed in-code.
Note: The base_scenario_name must be in each file read and combined by this tool, and entered in the general_inputs.csv file (1 MPG Standards is the default). However, since it's in each file,
it will be removed so that combining things doesn't double-count (multiple-count?) that scenario. In this tool, the total social costs and benefits are calculated relative to the
base_social_scenario entered in the general_inputs.csv file (2020hold is the default). This allows for any two scenarios to be compared since the total social cost and benefit results for both
are calculated relative to a common scenario.
"""
def __init__(self):
self.run_model_years = [year for year in range(2020, 2030)] # only used for the model year lifetime reports
self.summary_start_year = 2020
set_paths = SetPaths()
self.general_inputs = pd.read_csv(set_paths.path_inputs / 'general_inputs.csv', index_col=0).to_dict('index')
self.run_folder_filename = self.general_inputs['run_folder_filename']['value']
self.run_folders = pd.read_csv(set_paths.path_inputs / self.run_folder_filename, index_col=0).to_dict('index')
self.base_scenario_name = self.general_inputs['base_scenario_name']['value']
self.base_social_name = self.general_inputs['base_social_name']['value']
self.off_cycle_cost_per_credit = float(self.general_inputs['off_cycle_cost_per_credit']['value'])
self.model_runs = dict()
for k in self.run_folders.keys():
self.model_runs.update({k: [self.run_folders[k]['FWO_folder'], self.run_folders[k]['NFWO_folder'], self.run_folders[k]['YearShift']]})
# create full paths to folders contained in the model_runs dictionary
self.model_runs_path_dict = dict()
for k, v in self.model_runs.items():
if self.run_folder_filename.__contains__('primary'):
self.model_runs_path_dict[k] = [Path(set_paths.path_project / f'CAFE_model_runs/output/{v[0]}/reports-csv'),
Path(set_paths.path_project / f'CAFE_model_runs/output/{v[1]}/reports-csv'),
v[2]]
elif self.run_folder_filename.__contains__('sensitivities'):
self.model_runs_path_dict[k] = [Path(set_paths.path_project / f'CAFE_model_runs/sensitivities/output/{v[0]}/reports-csv'),
Path(set_paths.path_project / f'CAFE_model_runs/sensitivities/output/{v[1]}/reports-csv'),
v[2]]
else:
print('"run_folder_filename" in general_inputs.csv file must contain the keyword "primary" or "sensitivities"')
# set report names in case they are requested via the runtime settings
self.compliance_report_name = 'compliance_report'
self.costs_summary_report_name = 'annual_societal_costs_summary_report'
self.costs_report_name = 'annual_societal_costs_report'
self.effects_summary_report_name = 'annual_societal_effects_summary_report'
self.effects_report_name = 'annual_societal_effects_report'
self.tech_pens_report_name = 'technology_utilization_report'
self.vehicles_report_name = 'vehicles_report'
# lists for calcs in the compliance report
self.args_to_sum = ['Sales', 'Jobs',
'AC Efficiency Cost', 'AC Leakage Cost', 'Off-Cycle Cost', 'Tech Cost', 'Reg-Cost',
'HEV Cost', 'Tax Credit', 'Consumer WTP', 'Tech Burden',
'Credits Earned', 'Credits Out', 'Credits In', 'CO-2 Credits Earned', 'CO-2 Credits Out', 'CO-2 Credits In',
]
self.args_to_sales_weight = ['Average CW', 'Average FP',
'Avg AC Efficiency Cost', 'Avg AC Leakage Cost', 'Avg Off-Cycle Cost', 'Avg Tech Cost', 'Avg Fines', 'Avg Reg-Cost',
'Avg HEV Cost', 'Avg Tax Credit', 'Avg Consumer WTP', 'Avg Tech Burden',
]
self.args_to_sales_vmt_weight = ['CO-2 Standard', 'CO-2 Rating', 'CO-2 2cycle', 'CO-2 Credit Use', 'AC Efficiency', 'AC Leakage', 'Off-Cycle Credits']
# lists of metrics to exclude
self.effects_metrics_to_exclude = ['Admissions', 'Asthma', 'Attacks', 'Bronchitis',
'Premature', 'Respiratory', 'Restricted', 'Work Loss',
]
self.costs_metrics_to_exclude = ['Damage']
# lists of social costs and benefits
self.retail_fuel_expenditures = 'Retail Fuel Outlay'
self.fuel_tax_revenues = 'Fuel Tax Revenue'
self.social_cost_args = ['Tech Cost',
'Maint/Repair Cost',
'Congestion Costs',
'Noise Costs'
]
self.consumer_surplus_as_cost_args = ['Foregone Consumer Sales Surplus']
self.fatality_costs = 'Fatality Costs'
self.non_fatal_injury_costs = 'Non-Fatal Injury Costs'
self.non_fatal_crash_costs = 'Non-Fatal Crash Costs'
self.drive_value = 'Drive Value'
self.refueling_time_cost = 'Refueling Time Cost'
self.fatality_risk_value = 'Fatality Risk Value'
self.non_fatal_crash_risk_value = 'Non-Fatal Crash Risk Value'
self.petrol_market_externalities = 'Petroleum Market Externalities'
self.social_criteria_benefit_args = ['Criteria_Costs_3.0', 'Criteria_Costs_7.0']
self.social_scc_benefit_args = ['GHG_Costs_5.0', 'GHG_Costs_3.0', 'GHG_Costs_2.5', 'GHG_Costs_3.0_95']
# read criteria and scc cost factors
self.df = pd.read_csv(set_paths.path_inputs / self.general_inputs['criteria_cost_factors_filename']['value'])
self.key = pd.Series(zip(self.df['calendar_year'], zip(self.df['discount_rate'], self.df['reg_class'], self.df['fuel'])))
self.df.insert(0, 'id', self.key)
self.df.set_index('id', inplace=True)
self.criteria_cost_factors = self.df.to_dict('index')
self.scc_cost_factors = pd.read_csv(set_paths.path_inputs / self.general_inputs['scc_cost_factors_filename']['value'], index_col=0).to_dict('index')
self.energy_security_cost_factors \
= pd.read_excel(set_paths.path_inputs / self.general_inputs['energy_security_cost_factors_filename']['value'],
sheet_name='Average Premiums', index_col=0, skiprows=1, engine='openpyxl').to_dict('index')
# for discounting
self.costs_start = self.general_inputs['costs_start']['value']
self.discount_year = int(self.general_inputs['discount_year']['value'])
self.social_discount_rates = [0.03, 0.07]
self.criteria_discount_rates = [0.03, 0.07]
self.scc_discount_rates = [0.025, 0.03, 0.05]
# set values used in-code
self.vmt_car = float(self.general_inputs['vmt_car']['value'])
self.vmt_truck = float(self.general_inputs['vmt_truck']['value'])
self.kwh_per_gge = float(self.general_inputs['kwh_per_gge']['value'])
self.kwh_us_annual = float(self.general_inputs['kwh_us_annual']['value'])
self.gal_per_bbl = float(self.general_inputs['gal_per_bbl']['value'])
self.gallons_of_gasoline_us_annual = float(self.general_inputs['gallons_of_gasoline_us_annual']['value'])
self.bbl_oil_us_annual = float(self.general_inputs['bbl_oil_us_annual']['value'])
self.e0_in_retail_gasoline = float(self.general_inputs['e0_in_retail_gasoline']['value'])
self.imported_oil_share = float(self.general_inputs['imported_oil_share']['value'])
self.energy_density_ratio_e0 = float(self.general_inputs['energy_density_ratio_e0']['value'])
self.year_for_compares = int(self.general_inputs['year_for_compares']['value'])
self.grams_per_uston = float(self.general_inputs['grams_per_uston']['value'])
self.grams_per_metricton = float(self.general_inputs['grams_per_metricton']['value'])
self.metricton_per_uston = self.grams_per_uston / self.grams_per_metricton
self.uston_per_metricton = 1 / self.metricton_per_uston