<filename>mindinsight/lineagemgr/common/validator/model_parameter.py
# Copyright 2019 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""Define schema of model lineage input parameters."""
from marshmallow import Schema, fields, ValidationError, pre_load, validates
from marshmallow.validate import Range, OneOf

from mindinsight.lineagemgr.common.exceptions.error_code import LineageErrorMsg, \
    LineageErrors
from mindinsight.lineagemgr.common.exceptions.exceptions import \
    LineageParamTypeError, LineageParamValueError
from mindinsight.lineagemgr.common.log import logger
from mindinsight.lineagemgr.common.utils import enum_to_list
from mindinsight.lineagemgr.querier.querier import LineageType
from mindinsight.lineagemgr.querier.query_model import FIELD_MAPPING
from mindinsight.utils.exceptions import MindInsightException

try:
    from mindspore.dataset.engine import Dataset
    from mindspore.nn import Cell, Optimizer
    from mindspore.common.tensor import Tensor
    from mindspore.train.callback import _ListCallback
except (ImportError, ModuleNotFoundError):
    logger.error('MindSpore Not Found!')


class RunContextArgs(Schema):
    """Define the parameter schema for RunContext."""
    optimizer = fields.Function(allow_none=True)
    loss_fn = fields.Function(allow_none=True)
    net_outputs = fields.Function(allow_none=True)
    train_network = fields.Function(allow_none=True)
    train_dataset = fields.Function(allow_none=True)
    epoch_num = fields.Int(allow_none=True, validate=Range(min=1))
    batch_num = fields.Int(allow_none=True, validate=Range(min=0))
    cur_step_num = fields.Int(allow_none=True, validate=Range(min=0))
    parallel_mode = fields.Str(allow_none=True)
    device_number = fields.Int(allow_none=True, validate=Range(min=1))
    list_callback = fields.Function(allow_none=True)

    @pre_load
    def check_optimizer(self, data, **kwargs):
        optimizer = data.get("optimizer")
        if optimizer and not isinstance(optimizer, Optimizer):
            raise ValidationError({'optimizer': [
                "Parameter optimizer must be an instance of mindspore.nn.optim.Optimizer."
            ]})
        return data

    @pre_load
    def check_train_network(self, data, **kwargs):
        train_network = data.get("train_network")
        if train_network and not isinstance(train_network, Cell):
            raise ValidationError({'train_network': [
                "Parameter train_network must be an instance of mindspore.nn.Cell."]})
        return data

    @pre_load
    def check_train_dataset(self, data, **kwargs):
        train_dataset = data.get("train_dataset")
        if train_dataset and not isinstance(train_dataset, Dataset):
            raise ValidationError({'train_dataset': [
                "Parameter train_dataset must be an instance of "
                "mindspore.dataengine.datasets.Dataset"]})
        return data

    @pre_load
    def check_loss(self, data, **kwargs):
        net_outputs = data.get("net_outputs")
        if net_outputs and not isinstance(net_outputs, Tensor):
            raise ValidationError({'net_outpus': [
                "The parameter net_outputs is invalid. It should be a Tensor."
            ]})
        return data

    @pre_load
    def check_list_callback(self, data, **kwargs):
        list_callback = data.get("list_callback")
        if list_callback and not isinstance(list_callback, _ListCallback):
            raise ValidationError({'list_callback': [
                "Parameter list_callback must be an instance of "
                "mindspore.train.callback._ListCallback."
            ]})
        return data


class EvalParameter(Schema):
    """Define the parameter schema for Evaluation job."""
    valid_dataset = fields.Function(allow_none=True)
    metrics = fields.Dict(allow_none=True)

    @pre_load
    def check_valid_dataset(self, data, **kwargs):
        valid_dataset = data.get("valid_dataset")
        if valid_dataset and not isinstance(valid_dataset, Dataset):
            raise ValidationError({'valid_dataset': [
                "Parameter valid_dataset must be an instance of "
                "mindspore.dataengine.datasets.Dataset"]})
        return data


class SearchModelConditionParameter(Schema):
    """Define the search model condition parameter schema."""
    summary_dir = fields.Dict()
    loss_function = fields.Dict()
    train_dataset_path = fields.Dict()
    train_dataset_count = fields.Dict()
    test_dataset_path = fields.Dict()
    test_dataset_count = fields.Dict()
    network = fields.Dict()
    optimizer = fields.Dict()
    learning_rate = fields.Dict()
    epoch = fields.Dict()
    batch_size = fields.Dict()
    loss = fields.Dict()
    model_size = fields.Dict()
    limit = fields.Int(validate=lambda n: 0 < n <= 100)
    offset = fields.Int(validate=lambda n: 0 <= n <= 100000)
    sorted_name = fields.Str()
    sorted_type = fields.Str(allow_none=True)
    lineage_type = fields.Str(
        validate=OneOf(enum_to_list(LineageType)),
        allow_none=True
    )

    @staticmethod
    def check_dict_value_type(data, value_type):
        """Check dict value type and int scope."""
        for key, value in data.items():
            if key == "in":
                if not isinstance(value, (list, tuple)):
                    raise ValidationError("In operation's value must be list or tuple.")
            else:
                if not isinstance(value, value_type):
                    raise ValidationError("Wrong value type.")
                if value_type is int:
                    if value < 0 or value > pow(2, 63) - 1:
                        raise ValidationError("Int value should <= pow(2, 63) - 1.")
                    if isinstance(value, bool):
                        raise ValidationError("Wrong value type.")

    @staticmethod
    def check_param_value_type(data):
        """Check input param's value type."""
        for key, value in data.items():
            if key == "in":
                if not isinstance(value, (list, tuple)):
                    raise ValidationError("In operation's value must be list or tuple.")
            else:
                if isinstance(value, bool) or \
                        (not isinstance(value, float) and not isinstance(value, int)):
                    raise ValidationError("Wrong value type.")

    @validates("loss")
    def check_loss(self, data):
        """Check loss."""
        SearchModelConditionParameter.check_param_value_type(data)

    @validates("learning_rate")
    def check_learning_rate(self, data):
        """Check learning_rate."""
        SearchModelConditionParameter.check_param_value_type(data)

    @validates("loss_function")
    def check_loss_function(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("train_dataset_path")
    def check_train_dataset_path(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("train_dataset_count")
    def check_train_dataset_count(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("test_dataset_path")
    def check_test_dataset_path(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("test_dataset_count")
    def check_test_dataset_count(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("network")
    def check_network(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("optimizer")
    def check_optimizer(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("epoch")
    def check_epoch(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("batch_size")
    def check_batch_size(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("model_size")
    def check_model_size(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("summary_dir")
    def check_summary_dir(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @pre_load
    def check_comparision(self, data, **kwargs):
        """Check comparision for all parameters in schema."""
        for attr, condition in data.items():
            if attr in ["limit", "offset", "sorted_name", "sorted_type", "lineage_type"]:
                continue

            if not isinstance(attr, str):
                raise LineageParamValueError('The search attribute not supported.')

            if attr not in FIELD_MAPPING and not attr.startswith('metric_'):
                raise LineageParamValueError('The search attribute not supported.')

            if not isinstance(condition, dict):
                raise LineageParamTypeError("The search_condition element {} should be dict."
                                            .format(attr))

            for key in condition.keys():
                if key not in ["eq", "lt", "gt", "le", "ge", "in"]:
                    raise LineageParamValueError("The compare condition should be in "
                                                 "('eq', 'lt', 'gt', 'le', 'ge', 'in').")

            if attr.startswith('metric_'):
                if len(attr) == 7:
                    raise LineageParamValueError(
                        'The search attribute not supported.'
                    )
                try:
                    SearchModelConditionParameter.check_param_value_type(condition)
                except ValidationError:
                    raise MindInsightException(
                        error=LineageErrors.LINEAGE_PARAM_METRIC_ERROR,
                        message=LineageErrorMsg.LINEAGE_METRIC_ERROR.value.format(attr)
                    )
        return data

<filename>mindinsight/lineagemgr/common/validator/model_parameter.py
# Copyright 2019 Huawei Technologies Co., Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============================================================================
"""Define schema of model lineage input parameters."""
from marshmallow import Schema, fields, ValidationError, pre_load, validates
from marshmallow.validate import Range, OneOf

from mindinsight.lineagemgr.common.exceptions.error_code import LineageErrorMsg, \
    LineageErrors
from mindinsight.lineagemgr.common.exceptions.exceptions import \
    LineageParamTypeError, LineageParamValueError
from mindinsight.lineagemgr.common.log import logger
from mindinsight.lineagemgr.common.utils import enum_to_list
from mindinsight.lineagemgr.querier.querier import LineageType
from mindinsight.lineagemgr.querier.query_model import FIELD_MAPPING
from mindinsight.utils.exceptions import MindInsightException

try:
    from mindspore.dataset.engine import Dataset
    from mindspore.nn import Cell, Optimizer
    from mindspore.common.tensor import Tensor
    from mindspore.train.callback import _ListCallback
except (ImportError, ModuleNotFoundError):
    logger.error('MindSpore Not Found!')


class RunContextArgs(Schema):
    """Define the parameter schema for RunContext."""
    optimizer = fields.Function(allow_none=True)
    loss_fn = fields.Function(allow_none=True)
    net_outputs = fields.Function(allow_none=True)
    train_network = fields.Function(allow_none=True)
    train_dataset = fields.Function(allow_none=True)
    epoch_num = fields.Int(allow_none=True, validate=Range(min=1))
    batch_num = fields.Int(allow_none=True, validate=Range(min=0))
    cur_step_num = fields.Int(allow_none=True, validate=Range(min=0))
    parallel_mode = fields.Str(allow_none=True)
    device_number = fields.Int(allow_none=True, validate=Range(min=1))
    list_callback = fields.Function(allow_none=True)

    @pre_load
    def check_optimizer(self, data, **kwargs):
        optimizer = data.get("optimizer")
        if optimizer and not isinstance(optimizer, Optimizer):
            raise ValidationError({'optimizer': [
                "Parameter optimizer must be an instance of mindspore.nn.optim.Optimizer."
            ]})
        return data

    @pre_load
    def check_train_network(self, data, **kwargs):
        train_network = data.get("train_network")
        if train_network and not isinstance(train_network, Cell):
            raise ValidationError({'train_network': [
                "Parameter train_network must be an instance of mindspore.nn.Cell."]})
        return data

    @pre_load
    def check_train_dataset(self, data, **kwargs):
        train_dataset = data.get("train_dataset")
        if train_dataset and not isinstance(train_dataset, Dataset):
            raise ValidationError({'train_dataset': [
                "Parameter train_dataset must be an instance of "
                "mindspore.dataengine.datasets.Dataset"]})
        return data

    @pre_load
    def check_loss(self, data, **kwargs):
        net_outputs = data.get("net_outputs")
        if net_outputs and not isinstance(net_outputs, Tensor):
            raise ValidationError({'net_outpus': [
                "The parameter net_outputs is invalid. It should be a Tensor."
            ]})
        return data

    @pre_load
    def check_list_callback(self, data, **kwargs):
        list_callback = data.get("list_callback")
        if list_callback and not isinstance(list_callback, _ListCallback):
            raise ValidationError({'list_callback': [
                "Parameter list_callback must be an instance of "
                "mindspore.train.callback._ListCallback."
            ]})
        return data


class EvalParameter(Schema):
    """Define the parameter schema for Evaluation job."""
    valid_dataset = fields.Function(allow_none=True)
    metrics = fields.Dict(allow_none=True)

    @pre_load
    def check_valid_dataset(self, data, **kwargs):
        valid_dataset = data.get("valid_dataset")
        if valid_dataset and not isinstance(valid_dataset, Dataset):
            raise ValidationError({'valid_dataset': [
                "Parameter valid_dataset must be an instance of "
                "mindspore.dataengine.datasets.Dataset"]})
        return data


class SearchModelConditionParameter(Schema):
    """Define the search model condition parameter schema."""
    summary_dir = fields.Dict()
    loss_function = fields.Dict()
    train_dataset_path = fields.Dict()
    train_dataset_count = fields.Dict()
    test_dataset_path = fields.Dict()
    test_dataset_count = fields.Dict()
    network = fields.Dict()
    optimizer = fields.Dict()
    learning_rate = fields.Dict()
    epoch = fields.Dict()
    batch_size = fields.Dict()
    loss = fields.Dict()
    model_size = fields.Dict()
    limit = fields.Int(validate=lambda n: 0 < n <= 100)
    offset = fields.Int(validate=lambda n: 0 <= n <= 100000)
    sorted_name = fields.Str()
    sorted_type = fields.Str(allow_none=True)
    lineage_type = fields.Str(
        validate=OneOf(enum_to_list(LineageType)),
        allow_none=True
    )

    @staticmethod
    def check_dict_value_type(data, value_type):
        """Check dict value type and int scope."""
        for key, value in data.items():
            if key == "in":
                if not isinstance(value, (list, tuple)):
                    raise ValidationError("In operation's value must be list or tuple.")
            else:
                if not isinstance(value, value_type):
                    raise ValidationError("Wrong value type.")
                if value_type is int:
                    if value < 0 or value > pow(2, 63) - 1:
                        raise ValidationError("Int value should <= pow(2, 63) - 1.")
                    if isinstance(value, bool):
                        raise ValidationError("Wrong value type.")

    @staticmethod
    def check_param_value_type(data):
        """Check input param's value type."""
        for key, value in data.items():
            if key == "in":
                if not isinstance(value, (list, tuple)):
                    raise ValidationError("In operation's value must be list or tuple.")
            else:
                if isinstance(value, bool) or \
                        (not isinstance(value, float) and not isinstance(value, int)):
                    raise ValidationError("Wrong value type.")

    @validates("loss")
    def check_loss(self, data):
        """Check loss."""
        SearchModelConditionParameter.check_param_value_type(data)

    @validates("learning_rate")
    def check_learning_rate(self, data):
        """Check learning_rate."""
        SearchModelConditionParameter.check_param_value_type(data)

    @validates("loss_function")
    def check_loss_function(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("train_dataset_path")
    def check_train_dataset_path(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("train_dataset_count")
    def check_train_dataset_count(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("test_dataset_path")
    def check_test_dataset_path(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("test_dataset_count")
    def check_test_dataset_count(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("network")
    def check_network(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("optimizer")
    def check_optimizer(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @validates("epoch")
    def check_epoch(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("batch_size")
    def check_batch_size(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("model_size")
    def check_model_size(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, int)

    @validates("summary_dir")
    def check_summary_dir(self, data):
        SearchModelConditionParameter.check_dict_value_type(data, str)

    @pre_load
    def check_comparision(self, data, **kwargs):
        """Check comparision for all parameters in schema."""
        for attr, condition in data.items():
            if attr in ["limit", "offset", "sorted_name", "sorted_type", "lineage_type"]:
                continue

            if not isinstance(attr, str):
                raise LineageParamValueError('The search attribute not supported.')

            if attr not in FIELD_MAPPING and not attr.startswith('metric_'):
                raise LineageParamValueError('The search attribute not supported.')

            if not isinstance(condition, dict):
                raise LineageParamTypeError("The search_condition element {} should be dict."
                                            .format(attr))

            for key in condition.keys():
                if key not in ["eq", "lt", "gt", "le", "ge", "in"]:
                    raise LineageParamValueError("The compare condition should be in "
                                                 "('eq', 'lt', 'gt', 'le', 'ge', 'in').")

            if attr.startswith('metric_'):
                if len(attr) == 7:
                    raise LineageParamValueError(
                        'The search attribute not supported.'
                    )
                try:
                    SearchModelConditionParameter.check_param_value_type(condition)
                except ValidationError:
                    raise MindInsightException(
                        error=LineageErrors.LINEAGE_PARAM_METRIC_ERROR,
                        message=LineageErrorMsg.LINEAGE_METRIC_ERROR.value.format(attr)
                    )
        return data

# Copyright 2019 Huawei Technologies Co., Ltd # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # ============================================================================ Define schema of model lineage input parameters. Define the parameter schema for RunContext. Define the parameter schema for Evaluation job. Define the search model condition parameter schema. Check dict value type and int scope. Check input param's value type. Check loss. Check learning_rate. Check comparision for all parameters in schema.