from hestia_earth.calculation.abstract_hierarchy import Hierarchy
from hestia_earth.schema import EmissionJSONLD, SchemaType
from hestia_earth.calculation.emissions import EMISSIONS_MODELS_ORDER
from hestia_earth.calculation.utils import instantiate_model
from hestia_earth.calculation.emissions.backgroundmodel import Backgroungmodel
from hestia_earth.utils.api import download_hestia
from .version import VERSION
[docs]class Emissions_hierarchy(Hierarchy):
"""Model_Hierarchy class. This is an abstract class from which inherit all the hierarchies defined in Hestia. It
contains a set of common methods and instance variables that are typically used by every hierarchy. A hierarchy
defines the order to run the calculations in the models for a particular emission defined in Hestia. The order
is from higher tier to lower tier."""
def __init__(self):
"""Constructor method.
Arguments:
None
Returns instance."""
super().__init__()
self.emission = None
def build_emission(self):
term = self.term
self.emission = EmissionJSONLD()
self.emission.fields['@type'] = SchemaType.EMISSION.value
self.emission.fields['term'] = dict(self.get_reference(term))
self.emission.fields['description'] = self.term['description'] if self.term.get('description') else None
self.emission.fields['value'] = [self.round_to_significant(self.best_result, 10)]
self.emission.fields['method'] = dict(self.get_reference(self.chosen_model.term))
if self.best_tier == "background":
self.emission.fields['methodTier'] = self.best_tier
self.emission.fields['inputs'] = self.chosen_model.assessment_inputs
else:
self.emission.fields['methodTier'] = f"tier {self.best_tier}"
keys = ['term', 'description', 'value', 'methodTier', 'inputs']
self.emission.fields['recalculated'] = [key for key in keys if self.emission.fields[key] is not None and
self.emission.fields[key] != "" and self.emission.fields[key] != []]
self.emission.fields['recalculatedVersion'] = [VERSION] * len(self.emission.fields['recalculated'])
return dict(self.emission.to_dict())
def calculate_model(self, model, cycle):
condition = isinstance(model, Backgroungmodel)
calculate = getattr(model, f'calculate_{self.key}' if not condition else 'calculate_background')
self.best_result = calculate() if not condition else calculate(self.key)
self.best_tier = model.tier
def check_model(self, model, cycle):
condition = isinstance(model, Backgroungmodel)
check = getattr(model, f'check_{self.key}' if not condition else 'check')
return check(cycle)
[docs] def update_node(self, data):
"""Method that updates cycle representation emission.
Arguments:
data: It is a dataStore object with the representation with the emission to be updated
Returns boolean."""
data.update_representation_emission(self.build_emission())
[docs]class Emissions_field_hierarchy(Emissions_hierarchy):
"""Model_Hierarchy class. This is an abstract class from which inherit all the hierarchies defined in Hestia. It
contains a set of common methods and instance variables that are typically used by every hierarchy. A hierarchy
defines the order to run the calculations in the models for a particular emission defined in Hestia. The order
is from higher tier to lower tier."""
def __init__(self, emission):
"""Constructor method.
Arguments:
None
Returns instance."""
super().__init__()
prob_keys = {
'ch4ToAirInputsProductionNonFossil': 'ch4ToAirInputsProductionNon-Fossil',
'112TrichlorotrifluoroethaneToAirInputsProduction': '112-TrichlorotrifluoroethaneToAirInputsProduction',
'11DichlorotetrafluoroethaneToAirInputsProduction': '11-DichlorotetrafluoroethaneToAirInputsProduction',
'1112TetrafluoroethaneToAirInputsProduction': '1112-TetrafluoroethaneToAirInputsProduction'
}
self.key = emission
self.models = [instantiate_model(model, package='emissions') for model in EMISSIONS_MODELS_ORDER[self.key]]
self.term = download_hestia(self.key) if self.key not in prob_keys else download_hestia(prob_keys[self.key])
[docs] def process_hierarchy(self, data):
"""Method that runs the hierarchy through all the models for a given cycle.
Arguments:
cycle: This is a cycle in the representation required by the calculation engine.
(Different from the Schema representation)
Returns a file object."""
cycle = data.representation_cycle
for model in self.models:
if not self.check_model(model, cycle):
continue
else:
self.chosen_model = model
self.calculate_model(model, cycle)
self.update_node(data)
break