WTOAAM/health_monitoring/tests/test_mset.py

import unittest
import pandas as pd
import numpy as np
from pathlib import Path
from health_monitoring.services.mset_service import MSETService

class TestMSETService(unittest.TestCase):
    def setUp(self):
        """初始化测试数据"""
        self.service = MSETService()
        self.test_data = self.service.generate_test_data(500)

    def test_train_model_dimension(self):
        """测试模型训练的维度匹配"""
        gen_data = self.test_data['generator'].values
        self.service.train_model(gen_data)
        self.assertIsNotNone(self.service.matrixD)
        self.assertIsNotNone(self.service.normalDataBallTree)

    def test_subsystem_health_evaluation(self):
        """测试子系统健康度计算（含AHP权重）"""
        gen_data = self.test_data['generator']
        weights, score = self.service.evaluate_subsystem_health(gen_data)
        
        # 验证权重和评分有效性
        self.assertEqual(len(weights), 4, "发电机应有4个特征权重")
        self.assertTrue(0 <= score <= 100, f"评分异常: {score}")
        print(f"发电机健康度: {score:.2f}, 权重: {weights.round(2)}")

    def test_ahp_weight_calculation(self):
        """测试AHP权重计算逻辑"""
        # 模拟3个子系统的AHP矩阵
        matrix = np.array([[1, 2, 3], [1/2, 1, 2], [1/3, 1/2, 1]])
        eigenvalue, eigenvector = np.linalg.eig(matrix)
        max_idx = np.argmax(eigenvalue)
        weights = eigenvector[:, max_idx].real
        weights = weights / np.sum(weights)
        
        # 修正后的断言：将消息作为最后一个位置参数
        self.assertAlmostEqual(np.sum(weights), 1, 4, "AHP权重和应为1")
        print(f"AHP权重测试: {weights.round(2)}")

    def test_overall_health_with_ahp(self):
        """测试整体健康度（严格使用AHP矩阵）"""
        overall_score, subsys_scores = self.service.evaluate_turbine_health(self.test_data)
        
        self.assertTrue(0 <= overall_score <= 100, f"整体评分异常: {overall_score}")
        self.assertGreater(len(subsys_scores), 0, "至少存在一个子系统评分")
        print(f"整体健康度: {overall_score:.2f}, 子系统: {subsys_scores}")

if __name__ == "__main__":
    unittest.main()