DualFlow/models/Dynamic_anomaly_diagnosis/causal_structure.py

# -*- coding: utf-8 -*-
"""
causal_structure.py: 第二层 - 物理因果结构构建
该模块负责将专家知识库（如工艺层级划分、设备归属）转化为图结构（邻接矩阵）。
"""
import numpy as np
import pandas as pd
from config import config

class CausalStructureBuilder:
    def __init__(self, threshold_df):
        self.df = threshold_df
        self.sensor_list = self.df['ID'].tolist()
        self.id_to_idx = {name: i for i, name in enumerate(self.sensor_list)}
        self.num_sensors = len(self.sensor_list)
        self.col_layer = self._find_col_by_keyword(config.KEYWORD_LAYER)
        self.col_device = self._find_col_by_keyword(config.KEYWORD_DEVICE)

    def _find_col_by_keyword(self, keyword):
        if keyword in self.df.columns: return keyword
        for col in self.df.columns:
            if col.lower() == keyword.lower(): return col
        raise ValueError(f"错误: 未找到列名包含 '{keyword}' 的列")

    def build(self):
        """
        核心构建逻辑：基于规则生成传感器之间的有向连接关系（邻接矩阵）
        返回值包含：传感器列表、索引映射字典、邻接矩阵(adj_matrix)
        """
        # 初始化 N x N 的全零矩阵，0 表示无连接，1 表示有连接
        adj_matrix = np.zeros((self.num_sensors, self.num_sensors), dtype=int)
        nodes_info = {}
        
        # 1. 遍历解析所有节点的属性字典
        for _, row in self.df.iterrows():
            d_val = row[self.col_device]
            
            # 清洗设备名：处理空值、NaN 等异常输入
            dev_id = str(d_val).strip() if pd.notna(d_val) and str(d_val).strip() != '' else None
           
            # 清洗层级号：如果层级未定义或填写错误，赋予 -1 表示该节点不参与因果溯源
            try: l_val = int(row[self.col_layer])
            except: l_val = -1
            
            nodes_info[row['ID']] = {'layer': l_val, 'device': dev_id}
            
        count_edges = 0
        
        # 2. 嵌套循环对比每一对传感器，判断它们之间是否存在“因果通路”
        for i, src_name in enumerate(self.sensor_list):
            src_node = nodes_info.get(src_name)
            
            # 如果起始节点没有定义有效层级，则跳过
            if not src_node or src_node['layer'] == -1: continue
            src_l, src_d = src_node['layer'], src_node['device']
            
            for j, dst_name in enumerate(self.sensor_list):
                # 排除自身到自身的连接（防止图遍历时陷入死循环）
                if i == j: continue 
            
                dst_node = nodes_info.get(dst_name)
                
                # 如果目标节点没有定义有效层级，则跳过
                if not dst_node or dst_node['layer'] == -1: continue
                dst_l, dst_d = dst_node['layer'], dst_node['device']
                
                # ==================== (A) 层级约束 (Layer Constraint) ====================
                # 水厂工艺是从上游传导到下游的。溯源方向是由下到上。
                # dst_l == src_l: 允许在同层级（例如同一环节的不同传感器）平移寻找
                # dst_l == src_l - 1: 允许向上一级（上游环节）寻找原因。绝不允许越级或向下游找。
                is_layer_valid = (dst_l == src_l) or (dst_l == src_l - 1)
                if not is_layer_valid: continue
                    
                # ==================== (B) 设备约束 (Device Constraint) ====================
                # 如果两个传感器明确归属于不同的具体设备（如一个是 RO1 膜，一个是 RO2 膜），
                # 则判定它们之间物理隔离，不存在因果关系，切断连接。
                is_dev_valid = True
                if (src_d is not None) and (dst_d is not None):
                    if src_d != dst_d: is_dev_valid = False
                
                # 如果同时满足 层级约束 和 设备约束，则判定为有效通路
                if is_dev_valid:
                    adj_matrix[i, j] = 1
                    count_edges += 1
        return {"sensor_list": self.sensor_list, "sensor_to_idx": self.id_to_idx, "adj_matrix": adj_matrix}