Homec4science
Diffusion bilan_CO2 (master) Browse

bilan_CO2/scripts/Carbon_footprint.py
e83a81a8cf6dmaster

Last Change
Raw File

Carbon_footprint.py
View Options

import pandas as pd
import numpy as np
from functions import *
class Carbon_footprint():
# Constants for column names
YEAR = "Year"
CATEGORY = "Category"
VALUE = "Value"
CO2_col = 'Factor [kg CO2eq]'
def __init__(self, EPT_infile, input_file, nomenclature):
self.general_attributes = {
"author": "Hugo Cruz",
"author_mail": "hugo.cruz@epfl.ch",
}
self.nomenclature = nomenclature
self.bilan = pd.DataFrame()
self.data = pd.DataFrame()
self.category_name = {}
self.Theme_name = {}
self.ignore_column = []
#Translate this into english : Nomenclature utilisée, à gauche est le nom tel qu'il est dans le fichier d'entrée, à droite est le nom tel qu'il est écrit dans le fichier facteur emission
self.input_file = input_file
def import_database(self, path):
self.data = pd.read_excel(path)
def read_emission_factor(self, emission_factor_path):
factor = pd.read_excel(self.input_file, sheet_name = "Facteur emission",index_col=0)
factor.fillna(method='ffill', inplace=True)
factor.reset_index(inplace=True)
if "google" in emission_factor_path:
factor = pd.read_csv(emission_factor_path, encoding= 'cp1252', delimiter=';', decimal=",")
else:
factor = pd.read_excel(emission_factor_path, index_col=0)
CO2_col = self.CO2_col
#Standardize unit: MWh, m3, km
factor_df = factor.copy()
mask_kWh = factor_df["Unit"]=="kWh"
factor_df[CO2_col] = factor_df[CO2_col].where(~mask_kWh,factor_df[CO2_col]*1000)
factor_df["Unit"] = factor_df["Unit"].replace("kWh", "MWh")
mask_MJ = factor_df["Unit"]=="MJ"
factor_df[CO2_col] = factor_df[CO2_col].where(~mask_MJ,factor_df[CO2_col]*3600)
factor_df["Unit"] = factor_df["Unit"].replace("MJ", "MWh")
mask_m = factor_df["Unit"]=="m"
factor_df[CO2_col] = factor_df[CO2_col].where(~mask_m,factor_df[CO2_col]/1000)
factor_df["Unit"] = factor_df["Unit"].replace("m", "km")
self.emission_factor = factor_df
def calculate_emission(self):
manual_input_path = self.input_file
inputs = pd.read_excel(manual_input_path, sheet_name="Facteurs emissions")
inputs[["Theme", "Scope", "Data name mapping"]] = inputs[["Theme", "Scope", "Data name mapping"]].fillna(method='ffill')
raw_data = self.data.copy()
raw_data["Data description"] = "Valeur calculée"
NaN_values = raw_data[(raw_data["Value"].isna())&(~raw_data["CO2"].isna())]
raw_data = raw_data[~raw_data["Value"].isna()]
facteurs_emissions = inputs.copy()
raw_data["Year"] = self.data["Year"].astype(float).astype(str)
raw_data["Value"] = self.data["Value"].astype(float)
facteurs_emissions["Year"] = facteurs_emissions["Year"].astype(str)
facteurs_emissions.rename(columns={'Data name mapping': 'Category'}, inplace=True)
# Attempt to join on Category, Year, and Campus
facteurs_emissions_sub = facteurs_emissions[~facteurs_emissions["Campus"].isna()&(~(facteurs_emissions["Year"].isna()=="nan"))]
joined_data = pd.merge(raw_data, facteurs_emissions_sub, on=['Category', 'Year', 'Campus'], how='left', suffixes=('', '_drop'))
unmatched = joined_data[joined_data['Factor [kg CO2eq]'].isna()][raw_data.columns]
# Now, replace the index of unmatched with the original indices
#Filter with not Nan Year so it's won't merge based on common NaN Year
facteurs_emissions_sub = facteurs_emissions[(facteurs_emissions["Campus"].isna())&(~(facteurs_emissions["Year"]=="nan"))]
# Join on Category and Year if previous join had no match
if not unmatched.empty:
unmatched_index=unmatched.index
joined_data_year = pd.merge(unmatched, facteurs_emissions_sub, on=['Category', 'Year'], how='left', suffixes=('', '_drop'))
joined_data_year.index = unmatched_index
joined_data_year_not_nan = joined_data_year[~joined_data_year['Factor [kg CO2eq]'].isna()]
#Subsitute the original columns with the "_drop" columns
joined_data.update(joined_data_year_not_nan)
# Check for unmatched entries again
unmatched = joined_data[joined_data['Factor [kg CO2eq]'].isna()][raw_data.columns]
#Filter with not Nan campus so it's won't merge based on common NaN campus
facteurs_emissions_sub = facteurs_emissions[(~facteurs_emissions["Campus"].isna())&((facteurs_emissions["Year"]=="nan"))]
# Join on Category and Campus if previous join had no match
if not unmatched.empty:
unmatched_index=unmatched.index
joined_data_campus = pd.merge(unmatched, facteurs_emissions_sub, on=['Category', 'Campus'], how='left', suffixes=('', '_drop'))
joined_data_campus.index = unmatched_index
joined_data_campus_not_nan = joined_data_campus[~joined_data_campus['Factor [kg CO2eq]'].isna()]
joined_data.update(joined_data_campus_not_nan)
# Check for unmatched entries again
unmatched = joined_data[joined_data['Factor [kg CO2eq]'].isna()][raw_data.columns]
facteurs_emissions_sub = facteurs_emissions[(facteurs_emissions["Campus"].isna())&((facteurs_emissions["Year"]=="nan"))]
# Finally, join just on Category if none of the above conditions are met
if not unmatched.empty:
unmatched_index=unmatched.index
joined_data_category = pd.merge(unmatched, facteurs_emissions_sub, on='Category', how='left', suffixes=('', '_drop'))
joined_data_category.index = unmatched_index
joined_data_category_not_nan = joined_data_category[~joined_data_category['Factor [kg CO2eq]'].isna()]
joined_data.update(joined_data_category_not_nan)
# Drop duplicate columns resulted from the merges
joined_data = joined_data[[col for col in joined_data.columns if not col.endswith('_drop')]]
joined_data.drop_duplicates(subset = ['Category', 'Year', 'Campus'],inplace=True, keep="first")
unmatched = joined_data[joined_data['Factor [kg CO2eq]'].isna()]
print(unmatched["Category"].unique())
joined_data["CO2"] = joined_data["Factor [kg CO2eq]"]*joined_data["Value"]
joined_data["Year"] = joined_data["Year"].astype(float).astype(int)
#Add NaN_values to joined_data
joined_data = pd.concat([joined_data, NaN_values], ignore_index=True)
self.data_L2 = joined_data.copy()
def extrapolate(self, EPT_year, raw_pendular):
#Si une valeur est manquante pour une année, l'algorithme cherche une valeur dans le future et l'adapte selon la croissance
#Si c'est une valeur dans le future n'est pas disponible, alors elle va chercher dans le passé et adpate selon croissance
data = self.data_L2.copy()
years = data["Year"].unique()
years.sort()
years = years[:-1]
## Student plane interpolation
Category = self.nomenclature["Avion etudiants"]
Stud_years = [2016, 2017, 2018]
data = extrapolate_student_plane(data, EPT_year, Stud_years, Category)
for year in years:
for Theme in data["Theme"].unique():
mask = (data["Year"]==year)&(data["Theme"]==Theme)
#For each Theme where there is no data in this year, take the data from the nearest year
if data.loc[mask, "CO2"].isna().all():
#Find the nearest year where data["Theme"] is not empty
nearest_year = year
while (data.loc[(data["Year"]==nearest_year)&(data["Theme"]==Theme), "CO2"].isna().all()):
nearest_year += 1
if nearest_year>2100:
nearest_year = year
while (data.loc[(data["Year"]==nearest_year)&(data["Theme"]==Theme), "CO2"].isna().all()):
nearest_year -= 1
nearest_mask = (data["Year"]==nearest_year)&(data["Theme"]==Theme)
data_masked = data.loc[nearest_mask]
for category in data_masked["Category"].unique():
serie = pd.Series()
serie["Year"] = year
serie["Theme"] = Theme
serie["Category"] = category
if Theme == "Mobilité pendulaire":
#Produit en croix pour estimer la valeur de la mobilité pendulaire en fonction de l'évolution de la population
data_pendular = data.loc[(data["Year"]==nearest_year)&(data["Theme"]=="Mobilité pendulaire")]
self.raw_pendular = raw_pendular
pendular_year = self.raw_pendular.loc[(self.raw_pendular["Year"] == year)&(self.raw_pendular["Theme"]=="Mobilité pendulaire")&(self.raw_pendular["Category"]==category)]
pendular_year["Part_modale"] = pendular_year["Value"]/pendular_year["Value"].sum()
pendular_nearest = self.raw_pendular.loc[(self.raw_pendular["Year"] == nearest_year)&(self.raw_pendular["Theme"]=="Mobilité pendulaire")&(self.raw_pendular["Category"]==category)]
pendular_nearest["Part_modale"] = pendular_nearest["Value"]/pendular_nearest["Value"].sum()
mask = data_pendular["Category"]==category
pendular_EPT_a = pendular_year[pendular_year["Category"]==category]["Part_modale"].values*EPT_year[year]
pendular_CO2_n = data_pendular[mask]["CO2"].values
pendular_EPT_n = pendular_nearest[pendular_nearest["Category"]==category]["Part_modale"].values*EPT_year[nearest_year]
#Produit en croix:
serie["CO2"] = (pendular_EPT_a*pendular_CO2_n/pendular_EPT_n)[0]
else:
ratio_evolution = ((EPT_year.loc[year]-EPT_year.loc[nearest_year])/(EPT_year.loc[nearest_year]))
serie["CO2"] = data.loc[(data["Year"]==nearest_year)&(data["Theme"]==Theme)&(data["Category"]==category), "CO2"].sum()* (1+ratio_evolution)
serie["Value"] = data.loc[(data["Year"]==nearest_year)&(data["Theme"]==Theme)&(data["Category"]==category), "Value"].sum()*(1+ratio_evolution)
serie["Data description"] = "Estimé"
serie["Scope"] = data.loc[(data["Year"]==nearest_year)&(data["Theme"]==Theme), "Scope"].median()
serie["Campus"] = "EPFL"
serie["Category"] = Theme
data = data.append(serie, ignore_index=True)
self.data_L3A = data
def objectives(self, year_reference=2022):
data_L3A = self.data_L3A.copy()
#Objectif 2024
self.df_objectifs_2024 = objectives_2024(data_L3A, year_reference)
self.year_reference = year_reference
#Objectif 2030
df = pd.DataFrame()
df["Theme"] = data_L3A["Theme"].unique()
df["Year"] = 2030
df["Campus"] = "EPFL"
df["Scope"] = np.nan
#Definition des objectifs 2030:
mask = (data_L3A["Theme"] == "Electricite")|(data_L3A["Theme"] == "Mazout")|(data_L3A["Theme"]=="CAD")|(data_L3A["Theme"] == "Gaz")
Energy_2006 = data_L3A.loc[(data_L3A["Year"]==2006)&(mask)].sum()
Objective_2030_energy = Energy_2006["CO2"]*0.5
gaz_2030 = data_L3A.loc[(data_L3A["Year"]==2006)&(data_L3A["Theme"] == "Gaz")]["CO2"]*0.5 #Assume que le gaz en 2030 est la moitié de 2006
elec_2030 = Objective_2030_energy-gaz_2030
df.loc[df["Theme"]=="Electricite", "CO2"] = elec_2030.squeeze()
df.loc[df["Theme"]=="Electricite", "Objectif"] = "-50% energie par rapport à 2006"
df.loc[df["Theme"]=="Gaz", "CO2"] = gaz_2030.squeeze()
df.loc[df["Theme"]=="Gaz", "Objectif"] = "-50% energie par rapport à 2006"
df.loc[df["Theme"]=="Mazout", "CO2"] = 0
df.loc[df["Theme"]=="Mazout", "Objectif"] = "Remplacement par Electricite"
df.loc[df["Theme"]=="CAD", "CO2"] = 0
df.loc[df["Theme"]=="CAD", "Objectif"] = "Remplacement par Electricite"
mask_voyages_prof = (data_L3A["Theme"]=="Voyages professionnels")
trip_2019 = data_L3A.loc[(data_L3A["Year"]==2019)&(mask_voyages_prof)]["CO2"].sum()
Objective_2030_trip = trip_2019*0.7
df.loc[df["Theme"]=="Voyages professionnels", "CO2"] = Objective_2030_trip
df.loc[df["Theme"]=="Voyages professionnels", "Objectif"] = "-30% par rapport à 2019"
Pendulaire_2019 = data_L3A.loc[(data_L3A["Year"]==2019)&(data_L3A["Theme"]=="Mobilité pendulaire")]["CO2"].sum()
Objective_2030_pendulaire = Pendulaire_2019*0.7
df.loc[df["Theme"]=="Mobilité pendulaire", "CO2"] = Objective_2030_pendulaire
df.loc[df["Theme"]=="Mobilité pendulaire", "Objectif"] = "-30% par rapport à 2019"
Alimentation_2019 = data_L3A.loc[(data_L3A["Year"]==2019)&(data_L3A["Theme"]=="Alimentation")]["CO2"].sum()
Objective_2030_alimentation = Alimentation_2019*0.6
df.loc[df["Theme"]=="Alimentation", "CO2"] = Objective_2030_alimentation
df.loc[df["Theme"]=="Alimentation", "Objectif"] = "-40% par rapport à 2019"
df.loc[df["Theme"]=="Numérique (fabrication)", "CO2"] = 4829e3
df.loc[df["Theme"]=="Numérique (fabrication)", "Objectif"] = "Malgré la croissance, maintenir l'impact à 2019"
df.loc[df["Theme"]==self.nomenclature["Achats"], "CO2"] = 42000e3
df.loc[df["Theme"]==self.nomenclature["Achats"], "Objectif"] = "Malgré la croissance, maintenir l'impact à 2019"
df.loc[df["Theme"]=="Dechets", "CO2"] = data_L3A.loc[(data_L3A["Year"]==2019)&(data_L3A["Theme"]=="Dechets")]["CO2"].sum()
df["Data description"] = "Objectifs 2030 (confédération+EPFL)"
df["Category"] = df["Theme"]
for Theme in df["Theme"].unique():
df.loc[df["Theme"]==Theme, "Scope"] = data_L3A[data_L3A["Theme"] == Theme]["Scope"].iloc[0]
self.df_2030 = df
def interpolation_objectifs(self):
data_L3A = self.data_L3A.copy()
df_2030 = self.df_2030
years = np.arange(data_L3A["Year"].max(),2030)
for Theme in df_2030["Theme"].unique():
for year in years:
df_temp = pd.Series()
df_temp["Theme"] = Theme
df_temp["Year"] = year
df_temp["Campus"] = "EPFL"
df_temp["Category"] = df_temp["Theme"]
df_temp["Data description"] = "Objectifs 2030 (confédération+EPFL)"
df_temp["Scope"] = data_L3A[data_L3A["Theme"] == Theme]["Scope"].iloc[0]
mask = (data_L3A["Year"]==data_L3A["Year"].max())&(data_L3A["Theme"]==Theme)
if mask.sum():
df_temp["CO2"] = np.interp(year, [data_L3A["Year"].max(), 2030], [data_L3A.loc[mask, "CO2"].sum(), df_2030.loc[(df_2030["Theme"]==Theme)&(df_2030["Year"]==2030), "CO2"].sum()])
if year != data_L3A["Year"].max():
df_2030 = df_2030.append(df_temp, ignore_index=True)
else:
#If this data is not available for a specific theme, we take the value of the last year
df_temp["CO2"] = np.interp(year, [data_L3A["Year"].max(), 2030], [data_L3A.loc[(data_L3A["Year"]==data_L3A["Year"].max()-1)&(data_L3A["Theme"]==Theme), "CO2"].iloc[0], df_2030.loc[(df_2030["Theme"]==Theme)&(df_2030["Year"]==2030), "CO2"].sum()])
df_2030 = df_2030.append(df_temp, ignore_index=True)
data_L3B = pd.concat([data_L3A, df_2030, self.df_objectifs_2024], ignore_index=True)
Elect_emission_factor = data_L3B[(data_L3B["Theme"]=="Electricite")&(data_L3B["Year"]==self.year_reference)][self.CO2_col].mean()
nan_mask = (data_L3B["Value"].isna())&(data_L3B["Category"] == "Electricite")
data_L3B.loc[nan_mask, "Value"] = data_L3B[nan_mask]["CO2"].iloc[0]/Elect_emission_factor
self.data_L3B = data_L3B.copy()
c4science · Help