import numpy as np
from hestia_earth.calculation.abstract_model import Model
from hestia_earth.calculation.data.constants.n2o import N2O_FACTORS_BY_CLIMATE_ZONE
from hestia_earth.calculation.data.constants.n2o import N2O_FACTORS_BY_CROP
from hestia_earth.calculation.data.constants.generic import ATOMIC_WEIGHT_CONVERSIONS
from hestia_earth.utils.api import download_hestia
from hestia_earth.calculation.utils import\
most_relevant_measurement, residue_nitrogen, summation, format_lookup, primary_product
MODEL_KEY = 'n2OToAirFertilizerAndExcretaDirectStehfestBouwman2006'
[docs]class N2OToAirFertilizerAndExcretadirectStehfestBouwman2006(Model):
def __init__(self):
# Define model tier
self.tier = 2
self.term = download_hestia(MODEL_KEY)
# Define model requirements
self.clay = None
self.sand = None
self.organicCarbonContent = None
self.soilPh = None
self.eco_ClimateZone = None
self.crop_grouping = None
self.inorgN_total = None
self.orgN_total = None
self.excretaN_total = None
self.res_nitrogen = None
self.N_total = None
# Instantiate variables
self.n2OToAirInorganicFertilizerDirect = None
self.n2OToAirOrganicFertilizerDirect = None
self.n2OToAirExcretaDirect = None
self.n2OToAirCropResidueDecompositionDirect = None
self.n2OToAirAllOriginsDirect = None
# Define model constants
self.EF_CRes_N_N2O = 0.015711073034911115
self.climate_emissions = N2O_FACTORS_BY_CLIMATE_ZONE
self.crop_n2o_n = N2O_FACTORS_BY_CROP
self.conv_n2on_n2o = ATOMIC_WEIGHT_CONVERSIONS['Conv_Mol_N2ON_N2O']
self.grouping_lookup = format_lookup('crop.csv', 'cropgroupingstehfestbouwman')
def calculate_n2OToAirInorganicFertilizerDirect(self):
self.n2OToAirInorganicFertilizerDirect = \
self.calculate_n2OToAirAllOriginsDirect() * self.inorgN_total / self.N_total
return self.n2OToAirInorganicFertilizerDirect
def calculate_n2OToAirOrganicFertilizerDirect(self):
self.n2OToAirOrganicFertilizerDirect = \
self.calculate_n2OToAirAllOriginsDirect() * self.orgN_total / self.N_total
return self.n2OToAirOrganicFertilizerDirect
def calculate_n2OToAirExcretaDirect(self):
self.n2OToAirExcretaDirect = \
self.calculate_n2OToAirAllOriginsDirect() * self.excretaN_total / self.N_total
return self.n2OToAirExcretaDirect
def calculate_n2OToAirCropResidueDecompositionDirect(self):
self.n2OToAirCropResidueDecompositionDirect = self.res_nitrogen * self.EF_CRes_N_N2O
return self.n2OToAirCropResidueDecompositionDirect
def calculate_n2OToAirAllOriginsDirect(self):
self.n2OToAirAllOriginsDirect = np.amin(
((0.072 * self.N_total,
np.exp(
0.475 +
0.0038 * self.N_total + (0 if self.organicCarbonContent/100 < 0.01
else 0.0526 if self.organicCarbonContent/100 <= 0.03 else 0.6334) +
(0 if self.soilPh < 5.5 else -0.4836 if self.soilPh > 7.3 else -0.0693) +
(0 if self.sand/100 > 0.65 and self.clay/100 < 0.18 else -0.1528
if self.sand/100 < 0.65 and self.clay/100 < 0.35 else 0.4312) +
self.climate_emissions[str(self.eco_ClimateZone)] + self.crop_n2o_n[self.crop_grouping])
- np.exp(
0.475 +
(0 if self.organicCarbonContent/100 < 0.01 else 0.0526
if self.organicCarbonContent/100 <= 0.03 else 0.6334) +
(0 if self.soilPh < 5.5 else -0.4836 if self.soilPh > 7.3 else -0.0693) +
(0 if self.sand/100 > 0.65 and self.clay/100 < 0.18 else -0.1528
if self.sand/100 < 0.65 and self.clay/100 < 0.35
else 0.4312) + self.climate_emissions[str(self.eco_ClimateZone)] +
self.crop_n2o_n[self.crop_grouping])))) * self.conv_n2on_n2o
return self.n2OToAirAllOriginsDirect
def get_total_n(self, cycle):
inputs = cycle["inputs"].evalues()
N_total = [sum(input["value"]) if "units" in input["term"] and input["term"]["units"] == "kg N"
else 0 for input in inputs]
return N_total
def complete(self, completeness):
self.inorgN_total = 0 if self.inorgN_total == {} and completeness['fertilizer'] else self.inorgN_total
self.orgN_total = 0 if self.orgN_total == {} and completeness['fertilizer'] else self.orgN_total
self.excretaN_total = 0 if self.excretaN_total == {} and completeness['products'] else self.excretaN_total
self.res_nitrogen = 0 if self.res_nitrogen == {} and completeness['cropResidue'] else self.res_nitrogen
self.N_total = 0 if self.N_total == {} and completeness['fertilizer'] and completeness['products'] and\
completeness['cropResidue'] else self.N_total
def check_n2OToAirInorganicFertilizerDirect(self, cycle):
# Calculate total inorganic N fertilizer input
inputs = cycle["inputs"].evalues()
self.inorgN_total = summation([sum(input["value"]) if "units" in input["term"] and
input["term"]["units"] == "kg N" and
input["term"]["termType"] == "inorganicFertilizer" else 0
for input in inputs])
self.N_total = summation(self.get_total_n(cycle))
self.complete(cycle['dataCompleteness'])
return self.inorgN_total != {} and self.N_total != {} and self.check_n2OToAirAllOriginsDirect(cycle)
def check_n2OToAirOrganicFertilizerDirect(self, cycle):
# Calculate total organic N fertilizer input
inputs = cycle["inputs"].evalues()
self.orgN_total = summation([sum(input["value"]) if "units" in input["term"] and
input["term"]["units"] == "kg N" and
input["term"]["termType"] == "organicFertilizer" else 0
for input in inputs])
self.N_total = summation(self.get_total_n(cycle))
self.complete(cycle['dataCompleteness'])
return self.orgN_total != {} and self.N_total != {} and self.check_n2OToAirAllOriginsDirect(cycle)
def check_n2OToAirExcretaDirect(self, cycle):
# Calculate total Excreta product
inputs = cycle["inputs"].evalues()
self.excretaN_total = summation([sum(input["value"]) if "units" in input["term"] and
input["term"]["units"] == "kg N" and
input["term"]["termType"] == "animalProduct" else 0
for input in inputs])
self.N_total = summation(self.get_total_n(cycle))
self.complete(cycle['dataCompleteness'])
return self.excretaN_total != {} and self.N_total != {} and self.check_n2OToAirAllOriginsDirect(cycle)
def check_n2OToAirCropResidueDecompositionDirect(self, cycle):
# Calculate total Residue decomposition product
self.res_nitrogen = residue_nitrogen(cycle['products'])
self.complete(cycle['dataCompleteness'])
return self.res_nitrogen != {} and self.check_n2OToAirAllOriginsDirect(cycle)
def check_n2OToAirAllOriginsDirect(self, cycle):
self.clay = most_relevant_measurement(cycle['site']['measurements']['clayContent'], cycle['endDate'])
self.sand = most_relevant_measurement(cycle['site']['measurements']['sandContent'], cycle['endDate'])
self.organicCarbonContent = most_relevant_measurement(cycle['site']['measurements']['soilOrganicCarbonContent'],
cycle['endDate'])
self.soilPh = most_relevant_measurement(cycle['site']['measurements']['soilPh'], cycle['endDate'])
self.eco_ClimateZone = most_relevant_measurement(cycle['site']['measurements']['eco-ClimateZone'],
cycle['endDate'])
product = primary_product(cycle['products']) if cycle['products'] != {} else {}
self.crop_grouping = self.grouping_lookup.get(product['term']['@id'], {}) if product != {} else 'X'
self.N_total = summation(self.get_total_n(cycle))
self.complete(cycle['dataCompleteness'])
checkin = [self.clay, self.sand, self.organicCarbonContent, self.soilPh, self.eco_ClimateZone,
self.crop_grouping]
return all(v != {} for v in checkin) and self.N_total != {} and self.N_total > 0 and\
self.crop_grouping in self.crop_n2o_n