easy-ml/monitor/monitor_metric.py

# -*- coding: utf-8 -*-
"""
@author: yq
@time: 2024/11/1
@desc: 监控报告
"""
import threading
from typing import Dict

from commom import GeneralException
from entitys import MonitorConfigEntity, MetricFucResultEntity
from enums import ResultCodesEnum
from .report_generate import ReportWord, ReportExcel


class MonitorMetric():

    def __init__(self, monitor_config_path: str):
        self._monitor_config = MonitorConfigEntity.from_config(monitor_config_path)
        self.lock = threading.Lock()
        self._metric_value_dict: Dict[str, MetricFucResultEntity] = {}

    @property
    def metric_value_dict(self):
        return self._metric_value_dict

    def _update_metric_value_dict(self, key, value):
        with self.lock:
            self._metric_value_dict[key] = value

    #  TODO 多线程计算指标
    def calculate_metric(self, *args, **kwargs):
        metric_dict = self._monitor_config.metric_dict
        for metric_code, metric_clazz in metric_dict.items():
            metric_value = metric_clazz.calculate(*args, **kwargs)
            self._update_metric_value_dict(metric_code, metric_value)

    def generate_report(self):
        if ".docx" in self._monitor_config.template_path:
            ReportWord.generate_report(self._metric_value_dict, self._monitor_config.template_path)
        elif ".xlsx" in self._monitor_config.template_path:
            ReportExcel.generate_report(self._metric_value_dict, self._monitor_config.template_path)
        else:
            raise GeneralException(ResultCodesEnum.NOT_FOUND,
                                   message=f"模板【{self._monitor_config.template_path}】不能处理，请使用【.docx】或【.xlsx】模板")


if __name__ == "__main__":
    pass