Source code for oasislmf.computation.base

__all__ = [
    'ComputationStep',
]

import os
import pathlib
import logging
import json
import inspect
from ods_tools.oed import OedSource
from collections import OrderedDict

from ..utils.data import get_utctimestamp
from ..utils.exceptions import OasisException
from ..utils.inputs import update_config, str2bool, has_oasis_env, get_oasis_env, ArgumentTypeError
from oasislmf.utils.log import oasis_log




class ComputationStep:
    """
    "Abstract" Class for all Computation Step (ExposurePreAnalysis, GulCalc, ...)
    initialise the object with all specified param un step_param and sub- ComputationStep
    provide a generic interface to get the all those parameter definitions (get_params)

    the Run method must be implemented and contain le business execution logic.

    """



    step_params = []



    chained_commands = []


    def __init__(self, **kwargs):
        """
        initialise the ComputationStep objects:
         - do the basic check for required parameter (required)
         - provide default value if defined (default)
         - check path existence (pre_exist)
         - create necessary directories (is_dir, is_path)
        """


        self.logger = logging.getLogger(__name__)



        self.kwargs = kwargs

        self.logger.debug(f"{self.__class__.__name__}: " + json.dumps(self.kwargs, indent=4, default=str))


        self.run = oasis_log(self.run)


        for param in self.get_params():
            param_value = kwargs.get(param['name'])
            if param_value in [None, ""]:
                if param.get('required'):
                    raise OasisException(f"parameter {param['name']} is required "
                                         f"for Computation Step {self.__class__.__name__}")
                else:
                    param_value = param.get('default')

            if (getattr(param.get('type'), '__name__', None) == 'str2bool') and (not isinstance(param_value, bool)):
                try:
                    param_value = str2bool(param_value)
                except ArgumentTypeError:
                    raise OasisException(
                        f"The parameter '{param.get('name')}' has an invalid value '{param_value}' for boolean. Valid strings are (case insensitive):"
                        "\n  True:  ['yes', 'true', 't', 'y', '1']"
                        "\n  False: ['no', 'false', 'f', 'n', '0']"
                    )

            if (param.get('is_path')
                    and param_value is not None
                    and not isinstance(param_value, OedSource)):
                if param.get('pre_exist') and not os.path.exists(param_value):
                    raise OasisException(
                        f"The path {param_value} ({param['help']}) "
                        f"must exist for Computation Step {self.__class__.__name__}")
                else:
                    if param.get('is_dir'):
                        pathlib.Path(param_value).mkdir(parents=True, exist_ok=True)
                    else:
                        pathlib.Path(os.path.dirname(param_value)).mkdir(parents=True, exist_ok=True)
            setattr(self, param['name'], param_value)

    @classmethod


    def get_default_run_dir(cls):
        return os.path.join(os.getcwd(), 'runs', f'{cls.run_dir_key}-{get_utctimestamp(fmt="%Y%m%d%H%M%S")}')


    @classmethod


    def get_params(cls):
        """
        return all the params of the computation step defined in step_params
        and the params from the sub_computation step in chained_commands
        if two params have the same name, return the param definition of the first param found only
        this allow to overwrite the param definition of sub step if necessary.
        """
        params = []
        param_names = set()

        def all_params():
            for param in cls.step_params:
                yield param
            for command in cls.chained_commands:
                for param in command.get_params():
                    yield param

        for param in all_params():
            if param['name'] not in param_names:
                param_names.add(param['name'])
                params.append(param)

        return params


    @classmethod


    def get_arguments(cls, **kwargs):
        """
        Return a list of default arguments values for the functions parameters
        If given arg values in 'kwargs' these will override the defaults
        """
        func_args = {el['name']: el.get('default', None) for el in cls.get_params()}
        type_map = {el['name']: el.get('type', None) for el in cls.get_params()}

        func_kwargs = update_config(kwargs)
        env_override = str2bool(os.getenv('OASIS_ENV_OVERRIDE', default=False))

        for param in func_args:
            if env_override and has_oasis_env(param):
                func_args[param] = get_oasis_env(param, type_map[param])
            elif param in func_kwargs:
                func_args[param] = func_kwargs[param]
        return func_args


    @classmethod


    def get_signature(cls):
        """ Create a function signature based on the 'get_params()' return
        """
        try:
            # Create keyword params (without default values)
            params = ["{}=None".format(p.get('name')) for p in cls.get_params() if not p.get('default')]

            # Create keyword params (with default values)
            for p in [p for p in cls.get_params() if p.get('default')]:
                if isinstance(p.get('default'), str):
                    params.append("{}='{}'".format(p.get('name'), p.get('default')))
                elif isinstance(p.get('default'), dict):
                    params.append("{}=dict()".format(p.get('name'), p.get('default')))
                elif isinstance(p.get('default'), OrderedDict):
                    params.append("{}=OrderedDict()".format(p.get('name'), p.get('default')))
                else:
                    params.append("{}={}".format(p.get('name'), p.get('default')))

            exec('def func_sig({}): pass'.format(", ".join(params)))
            return inspect.signature(locals()['func_sig'])
        except Exception:
            # ignore any errors in signature creation and return blank
            return None


    def run(self):
        """method that will be call by all the interface to execute the computation step"""
        raise NotImplemented(f'Methode run must be implemented in {self.__class__.__name__}')