mindspore-mindinsight/mindinsight/profiler/common/validator/validate.py

# Copyright 2020 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.
# ============================================================================
"""Validate the profiler parameters."""
from mindinsight.profiler.common.exceptions.exceptions import ProfilerParamTypeErrorException, \
    ProfilerDeviceIdException, ProfilerOpTypeException, \
    ProfilerSortConditionException, ProfilerFilterConditionException, ProfilerGroupConditionException
from mindinsight.profiler.common.log import logger as log

AICORE_TYPE_COL = ["op_type", "execution_time", "execution_frequency", "precent"]
AICORE_DETAIL_COL = ["op_name", "op_type", "execution_time", "subgraph", "full_op_name"]
AICPU_COL = ["serial_number", "op_name", "total_time", "dispatch_time", "RunV2_start",
             "compute_start", "memcpy_start", "memcpy_end", "RunV2_end"]


def validate_condition(search_condition):
    """
    Verify the param in search_condition is valid or not.

    Args:
        search_condition (dict): The search condition.

    Raises:
        ProfilerParamTypeErrorException: If the type of the param in search_condition is invalid.
        ProfilerDeviceIdException: If the device_id param in search_condition is invalid.
        ProfilerOpTypeException: If the op_type param in search_condition is invalid.
        ProfilerGroupConditionException: If the group_condition param in search_condition is invalid.
        ProfilerSortConditionException: If the sort_condition param in search_condition is invalid.
        ProfilerFilterConditionException: If the filter_condition param in search_condition is invalid.
    """
    if not isinstance(search_condition, dict):
        log.error("Invalid search_condition type, it should be dict.")
        raise ProfilerParamTypeErrorException(
            "Invalid search_condition type, it should be dict.")

    if "device_id" in search_condition:
        device_id = search_condition.get("device_id")
        if not isinstance(device_id, str):
            raise ProfilerDeviceIdException("Invalid device_id type, it should be str.")

    if "op_type" in search_condition:
        op_type = search_condition.get("op_type")
        if op_type == "aicpu":
            search_scope = AICPU_COL
        elif op_type == "aicore_type":
            search_scope = AICORE_TYPE_COL
        elif op_type == "aicore_detail":
            search_scope = AICORE_DETAIL_COL
        else:
            raise ProfilerOpTypeException("The op_type must in ['aicpu', 'aicore_type', 'aicore_detail']")
    else:
        raise ProfilerOpTypeException("The op_type must in ['aicpu', 'aicore_type', 'aicore_detail']")

    if "group_condition" in search_condition:
        validata_group_condition(search_condition)

    if "sort_condition" in search_condition:
        validate_sort_condition(search_condition, search_scope)

    if "filter_condition" in search_condition:
        validate_filter_condition(search_condition)

def validata_group_condition(search_condition):
    """
    Verify the group_condition in search_condition is valid or not.

    Args:
        search_condition (dict): The search condition.

    Raises:
        ProfilerGroupConditionException: If the group_condition param in search_condition is invalid.
    """
    group_condition = search_condition.get("group_condition")
    if not isinstance(group_condition, dict):
        raise ProfilerGroupConditionException("The group condition must be dict.")
    if "limit" in group_condition:
        limit = group_condition.get("limit", 0)
        if isinstance(limit, bool) \
                or not isinstance(group_condition.get("limit"), int):
            log.error("The limit must be int.")
            raise ProfilerGroupConditionException("The limit must be int.")
        if limit < 1 or limit > 100:
            raise ProfilerGroupConditionException("The limit must in [1, 100].")

    if "offset" in group_condition:
        offset = group_condition.get("offset", 0)
        if isinstance(offset, bool) \
                or not isinstance(group_condition.get("offset"), int):
            log.error("The offset must be int.")
            raise ProfilerGroupConditionException("The offset must be int.")
        if offset < 0:
            raise ProfilerGroupConditionException("The offset must ge 0.")

        if offset > 1000000:
            raise ProfilerGroupConditionException("The offset must le 1000000.")


def validate_sort_condition(search_condition, search_scope):
    """
    Verify the sort_condition in search_condition is valid or not.

    Args:
        search_condition (dict): The search condition.
        search_scope (list): The search scope.

    Raises:
        ProfilerSortConditionException: If the sort_condition param in search_condition is invalid.
    """
    sort_condition = search_condition.get("sort_condition")
    if not isinstance(sort_condition, dict):
        raise ProfilerSortConditionException("The sort condition must be dict.")
    if "name" in sort_condition:
        sorted_name = sort_condition.get("name", "")
        err_msg = "The sorted_name must be in {}".format(search_scope)
        if not isinstance(sorted_name, str):
            log.error("Wrong sorted name type.")
            raise ProfilerSortConditionException("Wrong sorted name type.")
        if sorted_name not in search_scope:
            log.error(err_msg)
            raise ProfilerSortConditionException(err_msg)

    if "type" in sort_condition:
        sorted_type_param = ['ascending', 'descending']
        sorted_type = sort_condition.get("type")
        if sorted_type and sorted_type not in sorted_type_param:
            err_msg = "The sorted type must be ascending or descending."
            log.error(err_msg)
            raise ProfilerSortConditionException(err_msg)


def validate_filter_condition(search_condition):
    """
    Verify the filter_condition in search_condition is valid or not.

    Args:
        search_condition (dict): The search condition.

    Raises:
        ProfilerFilterConditionException: If the filter_condition param in search_condition is invalid.
    """
    def validate_op_filter_condition(op_condition):
        """
        Verify the op_condition in filter_condition is valid or not.

        Args:
            op_condition (dict): The op_condition in search_condition.

        Raises:
            ProfilerFilterConditionException: If the filter_condition param in search_condition is invalid.
        """
        if not isinstance(op_condition, dict):
            raise ProfilerFilterConditionException("Wrong op_type filter condition.")
        for key, value in op_condition.items():
            if not isinstance(key, str):
                raise ProfilerFilterConditionException("The filter key must be str")
            if not isinstance(value, list):
                raise ProfilerFilterConditionException("The filter value must be list")
            if key not in filter_key:
                raise ProfilerFilterConditionException("The filter key must in {}.".format(filter_key))
            for item in value:
                if not isinstance(item, str):
                    raise ProfilerFilterConditionException("The item in filter value must be str")

    filter_condition = search_condition.get("filter_condition")
    if not isinstance(filter_condition, dict):
        raise ProfilerFilterConditionException("The filter condition must be dict.")
    filter_key = ["in", "not_in", "partial_match_str_in"]
    if filter_condition:
        if "op_type" in filter_condition:
            op_type_condition = filter_condition.get("op_type")
            validate_op_filter_condition(op_type_condition)
        if "op_name" in filter_condition:
            op_name_condition = filter_condition.get("op_name")
            validate_op_filter_condition(op_name_condition)
        if "op_type" not in filter_condition and "op_name" not in filter_condition:
            raise ProfilerFilterConditionException("The key of filter_condition is not support")