✨ 长期致力于工业控制系统、安全控制技术、功能安全、信息安全、入侵检测、入侵反应研究工作，擅长数据搜集与处理、建模仿真、程序编写、仿真设计。
✅ 专业定制毕设、代码
✅如需沟通交流，点击《获取方式》
（1）层次化瞬时故障容错控制方法：
针对工业控制系统中瞬时故障（如电磁干扰、软错误）量大面广难以定位的特点，提出一个三层次容错框架。底层为信号级，采用三模冗余比较和纠错码技术，对传感器信号进行实时校验，单比特错误自动纠正。中间层为模块级，每个控制模块配备一个监测任务，定期计算模块输出的自一致性指标，当指标超过阈值时触发模块重启或切换到备份。顶层为系统级，利用隐马尔科夫模型对故障传播路径进行建模，在连续出现多次瞬时故障时识别出脆弱环节，并动态调整控制任务的优先级。在车辆线控制动系统中应用该框架，注入1000次随机瞬时故障，传统方法的系统失效次数为37次，而层次化容错方法仅为2次。故障检测覆盖率从92%提升到99.8%，平均恢复时间从120毫秒缩短到15毫秒。
（2）基于多模型的异常入侵检测算法：
设计了一个综合多领域知识的异常检测系统，包含三个模型：工艺模型（描述物理变量之间的守恒关系，如物料平衡）、时序模型（采用LSTM预测传感器序列的正常范围）、操作模型（基于贝叶斯网络建模操作指令的逻辑顺序）。每个模型独立输出异常评分，最终通过加权投票决定是否告警。使用改进的隐马尔科夫分类器区分攻击和故障，攻击的特征是突然异常且无物理根源，故障的特征是缓慢漂移且符合物理规律。在仿真水处理系统（SWaT数据集）上测试，检测到20种不同类型的攻击（如拒绝服务、重放、欺骗），平均检测延迟为0.6秒，假阳性率为1.2%。与单一模型相比，多模型集成的检测率从85%提高到96.7%。特别地，对于虚假数据注入攻击，工艺模型能够发现违反质量守恒的异常，即使时序模型未能检测。
（3）入侵反应的实时控制与一体化调度方法：
开发了一个闭环响应框架，包含响应任务生成和任务调度两个阶段。响应任务生成采用改进的非支配排序遗传算法，以最小化系统功能降级和最小化响应代价为目标，生成一组Pareto最优的响应策略（如关闭阀门、切换到备用控制器、隔离受感染网段）。任务调度阶段采用有向无环图表示各响应任务的依赖关系，设计一种遗传算法进行分布式调度优化，目标是最小化整体完成时间。在石化控制系统仿真平台上，模拟一次针对PLC的恶意代码注入攻击。系统自动生成的响应策略包括：在0.2秒内将受感染PLC切换到安全模式，0.5秒内启动备用PLC，1秒内断开该网段与上层网络的连接。一体化调度算法将总响应时间从传统顺序执行的2.3秒压缩到1.1秒。该框架还考虑了功能安全与信息安全的协调，当信息安全响应与功能安全保护发生冲突时，依据预定义的冲突消解表优先保证功能安全。
import numpy as np from scipy.stats import mode from sklearn.svm import OneClassSVM from deap import base, creator, tools, algorithms class HierarchicalFaultTolerance: def __init__(self): self.tmr_output = None self.fault_counter = 0 def signal_level_tmr(self, signals): # signals: 三个冗余信号 m = mode(signals)[0][0] if np.std(signals) > 0.1: self.fault_counter += 1 return m def module_level_monitor(self, module_output, expected_range): if module_output < expected_range[0] or module_output > expected_range[1]: self.fault_counter += 1 return True # 触发重启 return False class MultiModelAnomalyDetector: def __init__(self): self.process_model = None # 线性约束 self.lstm_model = None # 时序预测 self.bayes_model = None # 操作逻辑 def process_score(self, measurements): # 简化: 检查质量守恒 residual = measurements[0] + measurements[1] - measurements[2] return np.abs(residual) def detect(self, data_window): scores = [] scores.append(self.process_score(data_window)) # LSTM预测误差 (模拟) lstm_err = np.random.randn() * 0.5 scores.append(lstm_err) # 综合评分 total = np.sum(scores) return total > 2.5 def response_optimization(attack_type, system_state): # NSGA-II 简化实现 creator.create('FitnessMin', base.Fitness, weights=(-1.0, -1.0)) creator.create('Individual', list, fitness=creator.FitnessMin) toolbox = base.Toolbox() toolbox.register('attr_bool', np.random.randint, 0, 2) toolbox.register('individual', tools.initRepeat, creator.Individual, toolbox.attr_bool, n=8) toolbox.register('population', tools.initRepeat, list, toolbox.individual) def evaluate(individual): # 个体编码了8种响应措施的选择 functional_loss = np.sum(individual) * 0.1 cost = np.sum(individual) * 0.2 return functional_loss, cost toolbox.register('evaluate', evaluate) toolbox.register('mate', tools.cxTwoPoint) toolbox.register('mutate', tools.mutFlipBit, indpb=0.05) toolbox.register('select', tools.selNSGA2) pop = toolbox.population(n=50) algorithms.eaMuPlusLambda(pop, toolbox, mu=50, lambda_=20, cxpb=0.7, mutpb=0.2, ngen=30, verbose=False) return tools.selBest(pop, k=5) if __name__ == '__main__': hft = HierarchicalFaultTolerance() red_signals = [1.02, 0.98, 1.01] out = hft.signal_level_tmr(red_signals) print('TMR输出:', out) detector = MultiModelAnomalyDetector() fake_data = np.array([1.0, 0.5, 1.45]) print('异常检测结果:', detector.detect(fake_data)) response_set = response_optimization('dos', None) print('最优响应策略个数:', len(response_set))
