Health/status_config.py

# status_config.py
import pandas as pd
import os
from typing import Dict, List, Set

class StatusConfig:
    """状态测点配置管理器"""
    
    def __init__(self, status_csv_path: str = None):
        self.status_features = set()
        self.status_csv_path = status_csv_path
        self._load_status_features()
    
    def _load_status_features(self):
        """加载状态测点配置"""
        # 如果提供了CSV文件路径，从文件加载
        if self.status_csv_path and os.path.exists(self.status_csv_path):
            try:
                df = pd.read_csv(self.status_csv_path)
                if 'status_feature' in df.columns:
                    self.status_features = set(df['status_feature'].dropna().tolist())
                    print(f"加载 {len(self.status_features)} 个状态测点")
                return
            except Exception as e:
                print(f"加载状态测点CSV文件失败: {str(e)}")
        
        # 如果没有CSV文件或加载失败，使用内置的状态测点关键词
        status_keywords = [
            'sts', 'pump_act', 'sig', 'fault', 'conv_err', 'alarm', 
            'chain', 'estop', 'ups'
        ]
        self.status_features = set(status_keywords)
        print("使用内置状态测点关键词")
    
    def is_status_feature(self, feature_name: str) -> bool:
        """判断是否为状态测点"""
        feature_lower = feature_name.lower()
        return any(keyword in feature_lower for keyword in self.status_features)
    
    def filter_status_features(self, features: List[str]) -> List[str]:
        """过滤掉状态测点，只保留量测数据"""
        return [f for f in features if not self.is_status_feature(f)]
    
    def get_status_features(self, features: List[str]) -> List[str]:
        """获取状态测点列表"""
        return [f for f in features if self.is_status_feature(f)]