Source code for hestia_earth.calculation.impacts.leqScarcityWeightedWaterUseAware

from hestia_earth.calculation.abstract_model import Model
from hestia_earth.calculation.data.lookups import load_lookup
from hestia_earth.calculation.utils import summation
from hestia_earth.utils.lookup import get_table_value

MODEL_KEY = 'leqScarcityWeightedWaterUseAware'


class LeqScarcityWeightedWaterUseAware(Model):
    def __init__(self):

        super().__init__()

        # Model id
        self.id = MODEL_KEY

        # Define model requirements
        self.irrigation = None
        self.pyield = None
        self.seed = None
        self.econ = None
        self.awarecf = None
        self.emissonsResourceUse = None
        self.siteType = None

        # Instantiate variables
        self.scarcitywater = None

        # Define model coefficients
        self.characterization_factors = load_lookup("AWARE100.csv")

    def calculate(self):

        # Calculate scarcity water indicator
        self.scarcitywater =\
            (self.irrigation/self.pyield) * (self.seed / (self.seed + self.pyield)) * self.econ/100 * self.awarecf

        return self.scarcitywater

    def annual_characterization_factor(self, basin_id):

        if (self.siteType == 'cropland' or self.siteType == 'permanent pasture') and basin_id != {}:
            cf = get_table_value(self.characterization_factors, 'termid', basin_id, 'yr_irri')
        elif basin_id != {}:
            cf = get_table_value(self.characterization_factors, 'termid', basin_id, 'yr_nonirri')
        else:
            cf = None

        return float(cf) if cf is not None else {}

    def complete(self, completeness):

        self.irrigation = 0 if self.irrigation == {} and completeness['water'] else self.irrigation
        self.seed = 0 if self.seed == {} and completeness['other'] else self.seed

    def check(self, cycle, product, emissionsresourceuse):

        inputs = cycle['inputs'].evalues()
        self.irrigation = summation([sum(inp['value'])
                                    if 'units' in inp['term'] and inp['term']['termType'] == 'water'
                                    else 0 for inp in inputs])
        self.pyield = summation(product['value']) if 'value' in product else {}
        self.seed = summation(cycle['inputs']['seed']['value'])
        self.econ = product['economicValueShare'] if 'economicValueShare' in product else {}

        self.siteType = cycle['site']['siteType']
        awareWaterBasinId = cycle['site']['awareWaterBasinId']
        self.awarecf = self.annual_characterization_factor(f'b{awareWaterBasinId}')

        self.emissonsResourceUse = emissionsresourceuse

        self.complete(cycle['dataCompleteness'])

        return self.irrigation != {} and self.pyield != {} and self.pyield > 0 and self.seed != {} and\
            self.econ != {} and self.awarecf != {}