model_builder

Module for ModelBuilder class

!!! classes ModelBuilder

ModelBuilder (IModelBuilder)

Factory for creating models

Source code in workflow/model_builder.py

class ModelBuilder(IModelBuilder):
    """Factory for creating models"""

    def __init__(self, da_layer: IDataAccessLayer, logger: ILogger) -> None:
        self._logger = logger
        self._da_layer = da_layer

    def build_model(self, model_data: IModelData) -> IModel:
        """Creates a model based on model data.
        Current mapping works only for one dataset.

        Returns:
            IModel: instance of a model based on model data
        """

        self._logger.log_info("Creating rule-based model")

        datasets = [self._da_layer.read_input_dataset(ds) for ds in model_data.datasets]
        rules = list(ModelBuilder._create_rules(model_data.rules))

        mapping = model_data.datasets[0].mapping

        model: IModel = RuleBasedModel(
            datasets, rules, mapping, model_data.name, model_data.partition
        )

        return model

    @staticmethod
    def _create_rules(rule_data: List[IRuleData]) -> Iterable[IRule]:
        for rule_data_object in rule_data:
            yield ModelBuilder._create_rule(rule_data_object)

    @staticmethod
    def _set_default_fields(rule_data: IRuleData, rule: RuleBase):
        rule.description = rule_data.description
        rule.output_variable_name = rule_data.output_variable

    @staticmethod
    def _create_rule(rule_data: IRuleData) -> IRule:

        # from python >3.10 we can use match/case, better solution
        # until then disable pylint.
        # pylint: disable=too-many-branches
        if isinstance(rule_data, IMultiplyRuleData):
            rule = MultiplyRule(
                rule_data.name,
                [rule_data.input_variable],
                rule_data.multipliers,
                rule_data.date_range,
            )
        elif isinstance(rule_data, IDepthAverageRuleData):
            rule = DepthAverageRule(
                rule_data.name,
                rule_data.input_variables,
            )
        elif isinstance(rule_data, IFilterExtremesRuleData):
            rule = FilterExtremesRule(
                rule_data.name,
                rule_data.input_variables,
                rule_data.extreme_type,
                rule_data.distance,
                rule_data.time_scale,
                rule_data.mask,
            )
        elif isinstance(rule_data, ILayerFilterRuleData):
            rule = LayerFilterRule(
                rule_data.name,
                [rule_data.input_variable],
                rule_data.layer_number,
            )
        elif isinstance(rule_data, IAxisFilterRuleData):
            rule = AxisFilterRule(
                rule_data.name,
                [rule_data.input_variable],
                rule_data.element_index,
                rule_data.axis_name,
            )
        elif isinstance(rule_data, IStepFunctionRuleData):
            rule = StepFunctionRule(
                rule_data.name,
                rule_data.input_variable,
                rule_data.limits,
                rule_data.responses,
            )
        elif isinstance(rule_data, ITimeAggregationRuleData):
            rule = TimeAggregationRule(
                rule_data.name, [rule_data.input_variable], rule_data.operation
            )
            rule.settings.percentile_value = rule_data.percentile_value
            rule.settings.time_scale = rule_data.time_scale
        elif isinstance(rule_data, IRollingStatisticsRuleData):
            rule = RollingStatisticsRule(
                rule_data.name, [rule_data.input_variable], rule_data.operation
            )
            rule.settings.percentile_value = rule_data.percentile_value
            rule.settings.time_scale = rule_data.time_scale
            rule.period = rule_data.period
        elif isinstance(rule_data, ICombineResultsRuleData):
            rule = CombineResultsRule(
                rule_data.name,
                rule_data.input_variable_names,
                MultiArrayOperationType[rule_data.operation_type],
                rule_data.ignore_nan
            )
        elif isinstance(rule_data, IResponseCurveRuleData):
            rule = ResponseCurveRule(
                rule_data.name,
                rule_data.input_variable,
                rule_data.input_values,
                rule_data.output_values,
            )
        elif isinstance(rule_data, IFormulaRuleData):
            rule = FormulaRule(
                rule_data.name,
                rule_data.input_variable_names,
                rule_data.formula,
            )
        elif isinstance(rule_data, IClassificationRuleData):
            rule = ClassificationRule(
                rule_data.name, rule_data.input_variable_names, rule_data.criteria_table
            )
        else:
            error_str = (
                f"The rule type of rule '{rule_data.name}' is currently "
                "not implemented"
            )
            raise NotImplementedError(error_str)

        if isinstance(rule, RuleBase):
            ModelBuilder._set_default_fields(rule_data, rule)
        return rule

build_model(self, model_data)

Creates a model based on model data. Current mapping works only for one dataset.

Returns:

Type	Description
IModel	instance of a model based on model data

Source code in workflow/model_builder.py

def build_model(self, model_data: IModelData) -> IModel:
    """Creates a model based on model data.
    Current mapping works only for one dataset.

    Returns:
        IModel: instance of a model based on model data
    """

    self._logger.log_info("Creating rule-based model")

    datasets = [self._da_layer.read_input_dataset(ds) for ds in model_data.datasets]
    rules = list(ModelBuilder._create_rules(model_data.rules))

    mapping = model_data.datasets[0].mapping

    model: IModel = RuleBasedModel(
        datasets, rules, mapping, model_data.name, model_data.partition
    )

    return model