mindspore-mindinsight/mindinsight/lineagemgr/querier/querier.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.
# ============================================================================
"""This file is used to define lineage info querier."""
import enum
import functools
import operator
import os

from mindinsight.lineagemgr.common.exceptions.exceptions import \
    LineageParamTypeError, LineageSummaryAnalyzeException, \
    LineageEventNotExistException, LineageQuerierParamException, \
    LineageSummaryParseException, LineageEventFieldNotExistException
from mindinsight.lineagemgr.common.log import logger
from mindinsight.lineagemgr.common.utils import enum_to_list
from mindinsight.lineagemgr.querier.query_model import LineageObj, FIELD_MAPPING
from mindinsight.lineagemgr.summary.lineage_summary_analyzer import \
    LineageSummaryAnalyzer


@enum.unique
class ConditionParam(enum.Enum):
    """
    Filtering and sorting field names.

    `LIMIT` represents the number of lineage info per page. `OFFSET` represents
    page number. `SORTED_NAME` means to sort by this field. `SORTED_TYPE` means
    ascending or descending.
    """
    LIMIT = 'limit'
    OFFSET = 'offset'
    SORTED_NAME = 'sorted_name'
    SORTED_TYPE = 'sorted_type'
    LINEAGE_TYPE = 'lineage_type'

    @classmethod
    def is_condition_type(cls, value):
        """
        Judge that the input param is one of field names in the class.

        Args:
            value (str): The input field name.

        Returns:
            bool, `True` if the input field name in the class, else `False`.
        """
        return value in cls._value2member_map_


@enum.unique
class ExpressionType(enum.Enum):
    """
    Filter condition name definition.

    `EQ` means `==`. `LT` means `<`. `GT` means `>`. `LE` means `<=`. `GE` means
    `>=`. `IN` means filter value in the specified list.
    """
    EQ = 'eq'
    LT = 'lt'
    GT = 'gt'
    LE = 'le'
    GE = 'ge'
    IN = 'in'

    @classmethod
    def is_valid_exp(cls, key):
        """
        Judge that the input param is one of filter condition names in the class.

        Args:
            key (str): The input filter condition name.

        Returns:
            bool, `True` if the input filter condition name in the class,
            else `False`.
        """
        return key in cls._value2member_map_

    @classmethod
    def is_match(cls, except_key, except_value, actual_value):
        """
        Determine whether the value meets the expected requirement.

        Args:
            except_key (str): The expression key.
            except_value (Union[str, int, float, list, tuple]): The expected
                value.
            actual_value (Union[str, int, float]): The actual value.

        Returns:
            bool, `True` if the actual value meets the expected requirement,
            else `False`.
        """
        if actual_value is None and except_key in [cls.LT.value, cls.GT.value,
                                                   cls.LE.value, cls.GE.value]:
            return False

        try:
            if except_key == cls.IN.value:
                state = operator.contains(except_value, actual_value)
            else:
                state = getattr(operator, except_key)(actual_value, except_value)
        except TypeError:
            # actual_value can not compare with except_value
            return False
        return state


@enum.unique
class LineageFilterKey(enum.Enum):
    """Summary lineage information filter key."""
    METRIC = 'metric'
    HYPER_PARAM = 'hyper_parameters'
    ALGORITHM = 'algorithm'
    TRAIN_DATASET = 'train_dataset'
    VALID_DATASET = 'valid_dataset'
    MODEL = 'model'
    DATASET_GRAPH = 'dataset_graph'

    @classmethod
    def is_valid_filter_key(cls, key):
        """
        Judge that the input param is one of field names in the class.

        Args:
            key (str): The input field name.

        Returns:
            bool, `True` if the input field name in the class, else `False`.
        """
        return key in cls._value2member_map_

    @classmethod
    def get_key_list(cls):
        """
        Get the filter key name list.

        Returns:
            list[str], the filter key name list.
        """
        return [member.value for member in cls]


@enum.unique
class LineageType(enum.Enum):
    """Lineage search type."""
    DATASET = 'dataset'
    MODEL = 'model'


class Querier:
    """
    The querier of model lineage information.

    The class provides model lineage information query function. The information
    includes hyper parameters, train dataset, algorithm, model information,
    metric, valid dataset, etc.

    The class also provides search and sorting capabilities about model lineage
    information. You can search and sort by the specified condition.
    The condition explain in `ConditionParam` and `ExpressionType` class.
    See the method `filter_summary_lineage` for supported fields.

    Args:
        summary_path (Union[str, list[str]]): The single summary log path or
            a list of summary log path.

    Raises:
        LineageParamTypeError: If the input parameter type is invalid.
        LineageQuerierParamException: If the input parameter value is invalid.
        LineageSummaryParseException: If all summary logs parsing failed.
    """
    def __init__(self, summary_path):
        self._lineage_objects = []
        self._index_map = {}
        self._parse_failed_paths = []
        self._parse_summary_logs(summary_path)
        self._size = len(self._lineage_objects)

    def get_summary_lineage(self, summary_dir=None, filter_keys=None):
        """
        Get summary lineage information.

        If a summary dir is specified, the special summary lineage information
        will be found. If the summary dir is `None`, all summary lineage
        information will be found.

        Returns the content corresponding to the specified field in the filter
        key. The contents of the filter key include `metric`, `hyper_parameters`,
        `algorithm`, `train_dataset`, `valid_dataset` and `model`. You can
        specify multiple filter keys in the `filter_keys`. If the parameter is
        `None`, complete information will be returned.

        Args:
            summary_dir (Union[str, None]): Summary log dir. Default: None.
            filter_keys (Union[list[str], None]): Filter keys. Default: None.

        Returns:
            list[dict], summary lineage information.
        """
        self._parse_fail_summary_logs()

        if filter_keys is None:
            filter_keys = LineageFilterKey.get_key_list()
        else:
            for key in filter_keys:
                if not LineageFilterKey.is_valid_filter_key(key):
                    raise LineageQuerierParamException(
                        filter_keys, 'The filter key {} is invalid.'.format(key)
                    )

        if summary_dir is None:
            result = [
                item.get_summary_info(filter_keys) for item in self._lineage_objects
            ]
        else:
            index = self._index_map.get(summary_dir)
            if index is None:
                raise LineageQuerierParamException(
                    'summary_dir',
                    'Summary dir {} does not exist.'.format(summary_dir)
                )
            lineage_obj = self._lineage_objects[index]
            result = [lineage_obj.get_summary_info(filter_keys)]
        return result

    def filter_summary_lineage(self, condition=None):
        """
        Filter and sort lineage information based on the specified condition.

        See `ConditionType` and `ExpressionType` class for the rule of filtering
        and sorting. The filtering and sorting fields are defined in
        `FIELD_MAPPING` or prefixed with `metric/` or 'user_defined/'.

        If the condition is `None`, all model lineage information will be
        returned.

        Args:
            condition (Union[dict, None]): Filter and sort condition.
                Default: None.

        Returns:
            dict, filtered and sorted model lineage information.
        """
        def _filter(lineage_obj: LineageObj):
            for condition_key, condition_value in condition.items():
                if ConditionParam.is_condition_type(condition_key):
                    continue
                if self._is_valid_field(condition_key):
                    raise LineageQuerierParamException(
                        'condition',
                        'The field {} not supported'.format(condition_key)
                    )

                value = lineage_obj.get_value_by_key(condition_key)
                for exp_key, exp_value in condition_value.items():
                    if not ExpressionType.is_valid_exp(exp_key):
                        raise LineageQuerierParamException(
                            'condition',
                            'The expression {} not supported.'.format(exp_key)
                        )
                    if not ExpressionType.is_match(exp_key, exp_value, value):
                        return False
            return True

        def _cmp(obj1: LineageObj, obj2: LineageObj):
            value1 = obj1.get_value_by_key(sorted_name)
            value2 = obj2.get_value_by_key(sorted_name)

            if value1 is None and value2 is None:
                cmp_result = 0
            elif value1 is None:
                cmp_result = -1
            elif value2 is None:
                cmp_result = 1
            else:
                try:
                    cmp_result = (value1 > value2) - (value1 < value2)
                except TypeError:
                    type1 = type(value1).__name__
                    type2 = type(value2).__name__
                    cmp_result = (type1 > type2) - (type1 < type2)
            return cmp_result

        self._parse_fail_summary_logs()

        if condition is None:
            condition = {}
        results = list(filter(_filter, self._lineage_objects))

        if ConditionParam.SORTED_NAME.value in condition:
            sorted_name = condition.get(ConditionParam.SORTED_NAME.value)
            if self._is_valid_field(sorted_name):
                raise LineageQuerierParamException(
                    'condition',
                    'The sorted name {} not supported.'.format(sorted_name)
                )
            sorted_type = condition.get(ConditionParam.SORTED_TYPE.value)
            reverse = sorted_type == 'descending'
            results = sorted(
                results, key=functools.cmp_to_key(_cmp), reverse=reverse
            )

        offset_results = self._handle_limit_and_offset(condition, results)

        customized = self._organize_customized(offset_results)

        lineage_types = condition.get(ConditionParam.LINEAGE_TYPE.value)
        lineage_types = self._get_lineage_types(lineage_types)

        object_items = []
        for item in offset_results:
            lineage_object = dict()
            if LineageType.MODEL.value in lineage_types:
                lineage_object.update(item.to_model_lineage_dict())
            if LineageType.DATASET.value in lineage_types:
                lineage_object.update(item.to_dataset_lineage_dict())
            object_items.append(lineage_object)

        lineage_info = {
            'customized': customized,
            'object': object_items,
            'count': len(results)
        }

        return lineage_info

    def _organize_customized(self, offset_results):
        """Organize customized."""
        customized = dict()
        for offset_result in offset_results:
            for obj_name in ["metric", "user_defined"]:
                self._organize_customized_item(customized, offset_result, obj_name)

        # If types contain numbers and string, it will be "mixed".
        # If types contain "int" and "float", it will be "float".
        for key, value in customized.items():
            types = value["type"]
            if len(types) == 1:
                customized[key]["type"] = list(types)[0]
            elif types.issubset(["int", "float"]):
                customized[key]["type"] = "float"
            else:
                customized[key]["type"] = "mixed"
        return customized

    def _organize_customized_item(self, customized, offset_result, obj_name):
        """Organize customized item."""
        obj = getattr(offset_result, obj_name)
        require = bool(obj_name == "metric")
        if obj and isinstance(obj, dict):
            for key, value in obj.items():
                label = f'{obj_name}/{key}'
                current_type = type(value).__name__
                if customized.get(label) is None:
                    customized[label] = dict()
                    customized[label]["label"] = label
                    # user defined info is not displayed by default
                    customized[label]["required"] = require
                    customized[label]["type"] = set()
                customized[label]["type"].add(current_type)

    def _get_lineage_types(self, lineage_type_param):
        """
        Get lineage types.

        Args:
            lineage_type_param (dict): A dict contains "in" or "eq".

        Returns:
            list, lineage type.

        """
        # lineage_type_param is None or an empty dict
        if not lineage_type_param:
            return enum_to_list(LineageType)

        if lineage_type_param.get("in") is not None:
            return lineage_type_param.get("in")

        return [lineage_type_param.get("eq")]

    def _is_valid_field(self, field_name):
        """
        Check if field name is valid.

        Args:
            field_name (str): Field name.

        Returns:
            bool, `True` if the field name is valid, else `False`.
        """
        return field_name not in FIELD_MAPPING and \
               not field_name.startswith(('metric/', 'user_defined/'))

    def _handle_limit_and_offset(self, condition, result):
        """
        Handling the condition of `limit` and `offset`.

        Args:
            condition (dict): Filter and sort condition.
            result (list[LineageObj]): Filtered and sorted result.

        Returns:
            list[LineageObj], paginated result.
        """
        offset = 0
        limit = 10
        if ConditionParam.OFFSET.value in condition:
            offset = condition.get(ConditionParam.OFFSET.value)
        if ConditionParam.LIMIT.value in condition:
            limit = condition.get(ConditionParam.LIMIT.value)
        if ConditionParam.OFFSET.value not in condition \
                and ConditionParam.LIMIT.value not in condition:
            offset_result = result
        else:
            offset_result = result[offset * limit: limit * (offset + 1)]
        return offset_result

    def _parse_summary_logs(self, summary_path):
        """
        Parse summary logs.

        Args:
            summary_path (Union[str, list[str]]): The single summary log path or
                a list of summary log path.
        """
        if not summary_path:
            raise LineageQuerierParamException(
                'summary_path', 'The summary path is empty.'
            )
        if isinstance(summary_path, str):
            self._parse_summary_log(summary_path, 0)
        elif isinstance(summary_path, list):
            index = 0
            for path in summary_path:
                parse_result = self._parse_summary_log(path, index)
                if parse_result:
                    index += 1
        else:
            raise LineageParamTypeError('Summary path is not str or list.')

        if self._parse_failed_paths:
            logger.info('Parse failed paths: %s', str(self._parse_failed_paths))

        if not self._lineage_objects:
            raise LineageSummaryParseException()

    def _parse_summary_log(self, log_path, index: int, is_save_fail_path=True):
        """
        Parse the single summary log.

        Args:
            log_path (str): The single summary log path.
            index (int): TrainInfo instance index in the train info list.
            is_save_fail_path (bool): Set whether to save the failed summary
                path. Default: True.

        Returns:
            bool, `True` if parse summary log success, else `False`.
        """
        log_dir = os.path.dirname(log_path)
        try:
            lineage_info = LineageSummaryAnalyzer.get_summary_infos(log_path)
            user_defined_info = LineageSummaryAnalyzer.get_user_defined_info(log_path)
            lineage_obj = LineageObj(
                log_dir,
                train_lineage=lineage_info.train_lineage,
                evaluation_lineage=lineage_info.eval_lineage,
                dataset_graph=lineage_info.dataset_graph,
                user_defined_info=user_defined_info
            )
            self._lineage_objects.append(lineage_obj)
            self._add_dataset_mark()
            self._index_map[log_dir] = index
            return True
        except (LineageSummaryAnalyzeException,
                LineageEventNotExistException,
                LineageEventFieldNotExistException):
            if is_save_fail_path:
                self._parse_failed_paths.append(log_path)
            return False

    def _parse_fail_summary_logs(self):
        """Parse fail summary logs."""
        if self._parse_failed_paths:
            failed_paths = []
            for path in self._parse_failed_paths:
                parse_result = self._parse_summary_log(path, self._size, False)
                if parse_result:
                    self._size += 1
                else:
                    failed_paths.append(path)
            self._parse_failed_paths = failed_paths

    def _add_dataset_mark(self):
        """Add dataset mark into LineageObj."""
        # give a dataset mark for each dataset graph in lineage information
        marked_dataset_group = {'1': None}
        for lineage in self._lineage_objects:
            dataset_mark = '0'
            for dataset_graph_mark, marked_dataset_graph in marked_dataset_group.items():
                if marked_dataset_graph == lineage.dataset_graph:
                    dataset_mark = dataset_graph_mark
                    break
            # if no matched, add the new dataset graph into group
            if dataset_mark == '0':
                dataset_mark = str(int(max(marked_dataset_group.keys())) + 1)
                marked_dataset_group.update({
                    dataset_mark:
                        lineage.dataset_graph
                })
            lineage.dataset_mark = dataset_mark