from hestia_earth.calculation.abstract_model import Model
from hestia_earth.calculation.data.constants.p import C2_FACTORS_TILLAGE, SLOPE_RANGE
from hestia_earth.calculation.utils import most_relevant_measurement
from hestia_earth.utils.api import download_hestia
from hestia_earth.calculation.utils import summation, format_lookup, primary_product
MODEL_KEY = 'pErosionSalcaSchererPfister2015'
[docs]class PErosionSalcaSchererPfister2015(Model):
def __init__(self):
# Define model tier
self.tier = 1
self.term = download_hestia(MODEL_KEY)
# Define model requirements
self.nLAEnvironment = None
self.soilPhosphorusContent = None
self.water = None
self.wtype = None
self.product = None
self.erodibility = None
self.slope = None
self.slopeLength = None
self.country = None
self.management = None
self.pcorr = None
# Instantiate variables
self.A = None
self.R = None
self.pErosionAllOrigins = None
# Define model coefficients
self.c1 = format_lookup('crop.csv', 'p_ef_c1')
self.c2 = format_lookup('region.csv', 'ef_p_c2')
self.practiceFactors = format_lookup('region.csv', 'practice_factor')
self.tillage = C2_FACTORS_TILLAGE
self.srange = SLOPE_RANGE
def get_pcorr(self, slope):
pcorr = None
for element in self.srange:
if slope >= element[0] and slope < element[1]:
pcorr = element[2]
break
else:
pcorr = None
return pcorr
def get_management(self, cycle):
filter_management = ['fullTillage', 'reducedTillage', 'noTillage']
management = None
for practice in cycle['practices'].evalues():
if practice['term']['@id'] in filter_management:
management = practice['term']['@id']
return management
def calculate_R(self):
if (self.water) > 850:
R = 587.8 - 1.219 * (self.water) + \
0.004105 * (self.water) ** 2
else:
R = 0.0483 * (self.water) ** 1.61
return R
def calculate_A(self):
ppractice = self.practiceFactors[self.country]
A = self.R * self.erodibility * self.slopeLength * float(ppractice) * self.pcorr * float(self.c1[self.product])\
* (float(self.c2[self.country]) if self.product == 'Pasture' or self.management is None else
self.tillage[self.management])
return A
def calculate_pErosionAllOrigins(self):
self.R = self.calculate_R()
self.A = self.calculate_A()
self.pErosionAllOrigins = self.A * self.nLAEnvironment/100 * 2 * self.soilPhosphorusContent/1000
return self.pErosionAllOrigins
def complete(self, completeness):
self.water = 0 if self.water == {} and completeness['water'] else self.water
def check_pErosionAllOrigins(self, cycle):
self.nLAEnvironment =\
most_relevant_measurement(cycle['site']['measurements']['nutrientLossToAquaticEnvironment'],
cycle['endDate'])
self.soilPhosphorusContent =\
most_relevant_measurement(cycle['site']['measurements']['soilPhosphorusContent'], cycle['endDate'])
irrigation = summation(cycle['inputs']['irrigationPumpedGroundWater']['value'])
precipitation =\
most_relevant_measurement(cycle['site']['measurements']['rainfallAnnual'], cycle['endDate'])
self.water = summation([irrigation/10 if irrigation != {} else {}, precipitation])
product = primary_product(cycle['products']) if cycle['products'] != {} else {}
self.product = product['term']['@id'] if product != {} else {}
self.erodibility = most_relevant_measurement(cycle['site']['measurements']['erodibility'], cycle['endDate'])
self.slope = most_relevant_measurement(cycle['site']['measurements']['slope'], cycle['endDate'])
self.slopeLength = most_relevant_measurement(cycle['site']['measurements']['slopeLength'], cycle['endDate'])
self.country = cycle['site']['country']['@id']
self.management = self.get_management(cycle)
self.pcorr = self.get_pcorr(self.slope/100) if self.slope != {} else {}
self.complete(cycle['dataCompleteness'])
return self.nLAEnvironment != {} and self.soilPhosphorusContent != {} and self.water != {} and\
self.product != {} and self.product in self.c1 and self.erodibility != {} and self.slope != {} and\
self.slopeLength != {} and self.country != {} and self.country in self.c2 and self.country in\
self.practiceFactors and self.management != {} and self.pcorr is not None