Module for Application class
+!!! classes + Application
+ + + +
+Application
+
+
+
+Application for running command-line
+ +business/application.pyclass Application:
+ """Application for running command-line"""
+
+ # get version
+ APPLICATION_VERSION = read_version_number()
+ APPLICATION_NAME = "D-EcoImpact"
+ # separate version into major, minor and patch:
+ APPLICATION_VERSION_PARTS = list(map(int, APPLICATION_VERSION.split(".", 2)))
+
+ def __init__(
+ self,
+ logger: ILogger,
+ da_layer: IDataAccessLayer,
+ model_builder: IModelBuilder,
+ ):
+ """Creates an application based on provided logger, data-access layer
+ and model builder
+
+ Args:
+ logger (ILogger): Logger that takes care of logging
+ da_layer (IDataAccessLayer): data-access layer for reading/writing
+ model_builder (IModelBuilder): builder for creating a model based on
+ IModelData
+ """
+ self._logger = logger
+ self._da_layer = da_layer
+ self._model_builder = model_builder
+
+ def run(self, input_path: Path):
+ """Runs application
+
+ Args:
+ input_path (Path): path to input file
+ """
+
+ try:
+ # show application version
+ self._logger.log_info(f"Application version: {self.APPLICATION_VERSION}")
+
+ # read input file
+ model_data: IModelData = self._da_layer.read_input_file(input_path)
+ str_input_version = "".join([str(x) + "." for x in model_data.version])[:-1]
+ self._logger.log_info(f"Input file version: {str_input_version}")
+
+ # check version:
+ message = (
+ f"Application version {self.APPLICATION_VERSION} is older"
+ " than version from input file {str_input_version}"
+ )
+ # major version (app) should be equal or larger then input version --> error
+ if self.APPLICATION_VERSION_PARTS[0] < model_data.version[0]:
+ self._logger.log_error(message)
+ # minor version (app) should be equal or larger then input version --> warn
+ elif self.APPLICATION_VERSION_PARTS[1] < model_data.version[1]:
+ self._logger.log_warning(message)
+
+ # build model
+ for dataset in model_data.datasets:
+ input_files = self._da_layer.retrieve_file_names(dataset.path)
+ output_path_base = Path(model_data.output_path)
+ for key, file_name in input_files.items():
+ dataset.path = file_name
+ output_path = self._generate_output_path(output_path_base, key)
+
+ model_data.partition = key
+ model = self._model_builder.build_model(model_data)
+
+ # run model
+ _ModelRunner.run_model(model, self._logger)
+
+ # write output file
+ if model.status == _ModelStatus.FINALIZED:
+ settings = OutputFileSettings(
+ self.APPLICATION_NAME, self.APPLICATION_VERSION
+ )
+ settings.variables_to_save = model_data.output_variables
+
+ self._da_layer.write_output_file(
+ model.output_dataset, output_path, settings
+ )
+
+ except Exception as exc: # pylint: disable=broad-except
+ self._logger.log_error(f"Exiting application after error: {exc}")
+
+ def _generate_output_path(self, output_path_base, key):
+ if "*" in output_path_base.stem:
+ output_path = Path(str(output_path_base).replace("*", key))
+ else:
+ partition_part = ""
+ if key:
+ partition_part = f"_{key}"
+ output_path = Path.joinpath(
+ output_path_base.parent,
+ f"{output_path_base.stem}{partition_part}{output_path_base.suffix}",
+ )
+ return output_path
+__init__(self, logger: ILogger, da_layer: IDataAccessLayer, model_builder: IModelBuilder)
+
+
+ special
+
+
+Creates an application based on provided logger, data-access layer +and model builder
+ +Parameters:
+| Name | +Type | +Description | +Default | +
|---|---|---|---|
logger |
+ ILogger |
+ Logger that takes care of logging |
+ required | +
da_layer |
+ IDataAccessLayer |
+ data-access layer for reading/writing |
+ required | +
model_builder |
+ IModelBuilder |
+ builder for creating a model based on |
+ required | +
business/application.pydef __init__(
+ self,
+ logger: ILogger,
+ da_layer: IDataAccessLayer,
+ model_builder: IModelBuilder,
+):
+ """Creates an application based on provided logger, data-access layer
+ and model builder
+
+ Args:
+ logger (ILogger): Logger that takes care of logging
+ da_layer (IDataAccessLayer): data-access layer for reading/writing
+ model_builder (IModelBuilder): builder for creating a model based on
+ IModelData
+ """
+ self._logger = logger
+ self._da_layer = da_layer
+ self._model_builder = model_builder
+run(self, input_path: Path)
+
+
+Runs application
+ +Parameters:
+| Name | +Type | +Description | +Default | +
|---|---|---|---|
input_path |
+ Path |
+ path to input file |
+ required | +
business/application.pydef run(self, input_path: Path):
+ """Runs application
+
+ Args:
+ input_path (Path): path to input file
+ """
+
+ try:
+ # show application version
+ self._logger.log_info(f"Application version: {self.APPLICATION_VERSION}")
+
+ # read input file
+ model_data: IModelData = self._da_layer.read_input_file(input_path)
+ str_input_version = "".join([str(x) + "." for x in model_data.version])[:-1]
+ self._logger.log_info(f"Input file version: {str_input_version}")
+
+ # check version:
+ message = (
+ f"Application version {self.APPLICATION_VERSION} is older"
+ " than version from input file {str_input_version}"
+ )
+ # major version (app) should be equal or larger then input version --> error
+ if self.APPLICATION_VERSION_PARTS[0] < model_data.version[0]:
+ self._logger.log_error(message)
+ # minor version (app) should be equal or larger then input version --> warn
+ elif self.APPLICATION_VERSION_PARTS[1] < model_data.version[1]:
+ self._logger.log_warning(message)
+
+ # build model
+ for dataset in model_data.datasets:
+ input_files = self._da_layer.retrieve_file_names(dataset.path)
+ output_path_base = Path(model_data.output_path)
+ for key, file_name in input_files.items():
+ dataset.path = file_name
+ output_path = self._generate_output_path(output_path_base, key)
+
+ model_data.partition = key
+ model = self._model_builder.build_model(model_data)
+
+ # run model
+ _ModelRunner.run_model(model, self._logger)
+
+ # write output file
+ if model.status == _ModelStatus.FINALIZED:
+ settings = OutputFileSettings(
+ self.APPLICATION_NAME, self.APPLICATION_VERSION
+ )
+ settings.variables_to_save = model_data.output_variables
+
+ self._da_layer.write_output_file(
+ model.output_dataset, output_path, settings
+ )
+
+ except Exception as exc: # pylint: disable=broad-except
+ self._logger.log_error(f"Exiting application after error: {exc}")
+