k2health 0.0.7

K2data内部的设备健康分析模板工具包

K2Health

K2Health是K2Assets提供的设备健康模板开发包（以下简称SDK），它提供了一套分析流程，协助数据分析师利用从设备采集的时序数据构造数学模型，进而对设备健康状况进行评分。

一、安装

安装SDK最新版本：

pip install -U k2health

二、SDK概述

2.1 分析流程

在SDK中Pipeline类代表整个分析流程，也就是入口类。它由多个分析步骤组成，每个步骤通过构造方法接受参数配置，通过process函数实现处理逻辑。各步骤对应的类名如下：

• 数据清洗： DataCleaner
• 工况识别： ConditionPartitioner
• 模型训练： ModelTrainer （基于k2_health_rev2）
• 残差分析： TODO
• 健康评分： TODO
• 报警生成： TODO

2.2 参数表

用户通过参数表文件向分析流程各个步骤提供必要的配置参数，例如哪些测点作为要预测的测点、使用何种数学模型做预测等等。参数表是一个Excel格式的文件，它包含四个sheet，每个sheet代表一组参数用于不同的分析步骤：

• 测点配置：point_config，测点中英文名等信息
• 设备信息：device_config，设备中英文名称对照表
• 设备树：device_tree，设备组成关系
• 模型配置：model_config，健康模型测点影响关系和建模算法

我们期望在大多数情况下，用户只需要调整参数表中的配置，就可以实现对设备数据的分析并达到较好的效果。

文档中使用到的样例参数表： baoming_config.xlsx

2.3 设备数据

分析流程要处理的设备数据是DataFrame格式，要求包含timestamp时间戳列，其他数据列与参数表中的测点匹配。设备数据可以来自任何数据接口或离线文件。

文档中使用到的样例数据： baoming_sample_data.csv.zip

三、使用SDK

3.1 默认分析

默认方式是指按Pipeline类里所定义的顺序执行每个分析步骤，并且每个步骤的处理逻辑也是默认的，此时唯一能影响处理结果的只有参数表配置。示例代码如下：

from k2health.pipeline import *

config_file = "./health/sampledata/baoming_config.xlsx"

point_config = pd.read_excel(config_file, sheet_name='point_config')
x_col = ['motor_current', 'inlet_temperature', 'oil_temperature', 'total_inlet_flow', 
         'total_power']
y_col = ['motor_bearing_D_temperature', 'motor_bearing_D_vibX', 'motor_bearing_D_vibY',
         'motor_bearing_ND_temperature', 'motor_bearing_ND_vibX', 'motor_bearing_ND_vibY']
cleaner = DataCleaner(point_config, y_col, x_col)

device_config = pd.read_excel(config_file, sheet_name='device_config')
device_tree = pd.read_excel(config_file, sheet_name='device_tree')
partitioner = ConditionPartitioner(device_config, device_tree)

model_config_sheet = pd.read_excel(config_file, sheet_name='model_config')
trainer = ModelTrainer(y_col, model_config_sheet)

# 使用默认的数据处理器
pipeline = Pipeline(
    cleaner=cleaner,
    partitioner=partitioner,
    trainer=trainer
)

data = pd.read_csv("./sampledata/baoming_sample_data.csv")
pipeline.process(data)

使用上面代码训练得到的健康模型：

{
 'motor_bearing_D_temperature': {'default': {'model': Pipeline(steps=[('scaler', StandardScaler()),
                   ('regression', LinearRegression())]),
   'X': ['inlet_temperature', 'total_inlet_flow', 'oil_temperature'],
   'model_type': 'linear',
   'score': 0.9837478541653495,
   'sigma': 0.08700518226298681,
   'start_time': Timestamp('2022-02-01 00:00:00'),
   'end_time': Timestamp('2022-05-01 23:59:00'),
   'residue_mean': -5.998345055700141e-15}},
   
 'motor_bearing_D_vibX': {'default': {'model': Pipeline(steps=[('scaler', StandardScaler()),
                   ('regression', LinearRegression())]),
   'X': ['inlet_temperature', 'total_inlet_flow', 'motor_current'],
   'model_type': 'linear',
   'score': 0.34306575067257594,
   'sigma': 0.10422401013010404,
   'start_time': Timestamp('2022-02-01 00:00:00'),
   'end_time': Timestamp('2022-05-01 23:59:00'),
   'residue_mean': 1.05404310606428e-15}},
   
 ...
}

3.2 定制化分析

定制化方式是指在每个分析步骤基础上增加额外的处理逻辑，或者重写原有的处理逻辑，以满足特定的业务需求。此时用户需要开发自己的分析步骤类，并将它替换到分析流程里。

例如希望在默认的数据清洗完成后，额外对数据再做一次填充空值的处理。可以按下面的方式实现：

步骤1、定义一个继承自DataCleaner的子类CustomDataCleaner，实现process方法：

class CustomDataCleaner(DataCleaner):
    def process(self, data: DataFrame) -> DataFrame:
        # 先进行默认处理
        # 如果省略这一步，则相当于完全重写原有的处理逻辑
        data = super().process(data)  
        
        # 再进行定制化处理
        data = data.fillna(-1)
        return data

步骤2、在分析流程里引用CustomDataCleaner：

# 让Pipeline使用自定义的数据处理器
cleaner = CustomDataCleaner(point_config, y_col, x_col)
pipeline = Pipeline(
    cleaner=cleaner,
    partitioner=partitioner,
    trainer=trainer
)

# 其他代码与默认实现相同，略
...