from hestia_earth.utils.tools import current_time_ms
from .representation import DataStore
from .log import logger
[docs]class Engine():
"""Engine class. This is the engine abstract class for that orchestrates calculations for Hestia."""
def __init__(self):
"""Constructor method.
Arguments:
None
Returns instance."""
self.name = None
self.hierarchies = []
self.transformation_hierarchies = []
self.data = DataStore()
self.arguments = {'cycle': False, 'site': False, 'assessments': False}
[docs] def import_assessments(self, assessments):
"""Method that imports a jsonld file into the representation required to perform the Hestia calculations.
Arguments:
file: A jsonld file
Returns nothing."""
self.data.import_assessment(assessments)
[docs] def import_cycle(self, cycle, site=None):
"""Method that imports a jsonld file into the representation required to perform the Hestia calculations.
Arguments:
file: A jsonld file
Returns nothing."""
self.data.import_cycle(cycle, site)
def import_data(self, cycle=None, site=None, assessments=None):
self.arguments['cycle'] = True if cycle is not None else False
self.arguments['site'] = True if site is not None else False
self.arguments['assessments'] = True if assessments is not None else False
self.import_assessments(assessments)
if cycle is not None:
self.import_cycle(cycle, site)
def export_data(self):
if self.data.representation_cycle is not None and self.data.representation_assessment is None:
return self.export_cycle()
else:
return self.export_assessments()
[docs] def export_assessments(self):
"""Method that imports a jsonld file into the representation required to perform the Hestia calculations.
Arguments:
file: A jsonld file
Returns nothing."""
return self.data.export_assessment()
[docs] def export_cycle(self):
"""Method that exports the Engine representation into the Hestia Schema.
Arguments:
None
Returns a dict representing the recalculated cycle."""
if self.arguments['site']:
output = self.data.export_cycle()
del(output['site'])
return output
else:
return self.data.export_cycle()
def calculate_assessments(self):
self.assessment.process_hierarchy(self.data)
[docs] def calculate_emissions(self):
"""Method that orchestrates the execution of hierarchies of the engine.
Arguments:
None
Returns nothing."""
for hierarchy in self.hierarchies:
hierarchy.process_hierarchy(self.data)
def run(self):
now = current_time_ms()
if self.data.representation_cycle is not None and self.data.representation_assessment is None:
self.calculate_emissions()
if self.data.representation_assessment is not None:
self.calculate_assessments()
if self.data.representation_assessment is not None:
self.calculate_transformation_emissions()
logger.info('time=%s, unit=ms', current_time_ms() - now)