easy-ml/entitys/monitor_entity.py

# -*- coding: utf-8 -*-
"""
@author: yq
@time: 2024/11/1
@desc:  常用指标实体集合
"""
import json
import os
from typing import List, Dict

from commom import GeneralException
from enums import ResultCodesEnum
from metrics import f_get_metric_clazz_dict, MetricBase
from .metric_entity import MetricConfigEntity


class MonitorConfigEntity():

    def __init__(self, metric_config_list: List[MetricConfigEntity], template_path: str):
        self._template_path = template_path
        self._metric_clazz_dict = f_get_metric_clazz_dict()
        self._metric_dict: Dict[str, MetricBase] = self._init_metric(metric_config_list)

    @property
    def template_path(self):
        return self._template_path

    @property
    def metric_dict(self):
        return self._metric_dict

    def _init_metric(self, metric_config_list: List[MetricConfigEntity]) -> Dict[str, MetricBase]:
        metric_dict = {}
        for metric_config in metric_config_list:
            metric_func_name = metric_config.metric_func
            metric_code = metric_config.metric_code
            # 指标函数不存在
            if metric_func_name not in self._metric_clazz_dict.keys():
                raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"指标函数【{metric_func_name}】不存在")
            # 指标code不唯一
            if metric_code in metric_dict.keys():
                raise GeneralException(ResultCodesEnum.ILLEGAL_PARAMS, message=f"指标code【{metric_code}】不唯一")
            metric_clazz = self._metric_clazz_dict[metric_func_name]
            metric_dict[metric_code] = metric_clazz(*metric_config.args, **metric_config.kwargs)
        return metric_dict

    @staticmethod
    def from_config(config_path: str):
        """
        从配置文件生成实体类
        """
        if os.path.exists(config_path):
            with open(config_path, mode="r", encoding="utf-8") as f:
                j = json.loads(f.read())
        else:
            raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"指标监控配置文件【{config_path}】不存在")
        metric_config_list = j.get("metric_config_list", [])
        metric_config_list = [MetricConfigEntity(**i) for i in metric_config_list]
        j["metric_config_list"] = metric_config_list
        return MonitorConfigEntity(**j)


if __name__ == "__main__":
    pass