✨ 长期致力于船舶燃气轮机、仿真软件、系统仿真、分布式仿真、多软件协同研究工作，擅长数据搜集与处理、建模仿真、程序编写、仿真设计。
✅ 专业定制毕设、代码
✅ 如需沟通交流，点击《获取方式》

（1）基于两层接口封装的自治仿真模型重用架构：

为了解决不同仿真软件之间的模型互操作难题，设计了一种基于高层体系架构的分布式集成仿真平台。每个自治仿真模型（如Simulink中的压气机模型、Fluent中的燃烧室模型）被封装为一个联邦成员，封装包含两层接口：第一层为模型本地接口，负责调用仿真软件的内部应用编程接口进行模型初始化、单步执行和数据读写；第二层为高层体系架构接口，实现与运行时间基础设施的数据交互，包括声明发布/订购关系、时间推进和反射属性值。两层之间通过一个共享内存结构进行解耦，数据交换延迟平均为32微秒。针对某型三轴燃气轮机，将进气道的Fluent模型、燃烧室的化学动力学模型和涡轮的Simulink模型分别封装，成功构建了跨平台的联合仿真。在该架构下，模型重用率提高了60%，新模型的平均集成时间从原来的3人天缩短到0.5人天。

（2）基于参数变量的运行时间基础设施接口模块统一封装方法：

在Simulink环境中，由于回调函数中不能直接调用运行时间基础设施应用程序编程接口，本方法设计了一个S-Function模块，将多个不同功能的运行时间基础设施接口函数（如发布、订购、时间推进、反射值等）打包成一个参数变量结构体。该结构体包含函数句柄、数据缓冲区指针和状态标志位。在S-Function的初始化回调中，通过mex函数调用C++编写的动态链接库建立连接；在每个仿真步长中，通过Update和Output回调分别调用发布和订购函数。对于订购数据，设计了接收对象类实例属性反射值的算法，依据预先配置的订购信息表来匹配数据生产者和消费者，解决了联邦成员之间无法通过运行时间基础设施直接查询对方仿真步长的问题。该封装方法使得Simulink模型可以直接参与分布式仿真，而无需修改原有的模型方程。在包含8个联邦成员的分布式仿真中，时间同步精度达到0.1毫秒，丢包率低于0.01%。

（3）多学科协同与仿真缩放应用验证：

分别以COGAG装置和三轴燃气轮机为对象，进行了多学科耦合和仿真缩放验证。在COGAG装置中，将Simulink中的齿轮箱动力学模型与ADAMS中的多体动力学模型耦合，分析了两台燃气轮机并车时的扭振特性。仿真显示，当一台机组突然脱开时，输出轴的转矩波动峰值为额定转矩的2.3倍，持续0.18秒。联合仿真的结果与实测数据相比，峰值误差小于8%，证明了多学科耦合的准确性。在三轴燃气轮机仿真缩放实验中，将0维变比热系统模型与Fluent三维燃烧室模型耦合，通过分布式平台实现了区域分解：燃烧室区域用三维计算流体动力学求解，其余部件用零维集总参数模型。时间步长统一为0.001秒，但三维区域采用二次网格内迭代。结果表明，联合仿真的出口温度场分布与全三维仿真相差仅2.5%，而计算时间从全三维的7小时减少到1.2小时，加速比为5.8倍。该平台还支持任意增减联邦成员，扩展性强。

import numpy as np
import threading
import queue
from abc import ABC, abstractmethod

class HLAFederate(ABC):
    def __init__(self, name, step_size=0.001):
        self.name = name
        self.step = step_size
        self.send_queue = queue.Queue()
        self.recv_queue = queue.Queue()
        self.publish_handles = {}
        self.subscribe_handles = {}
    @abstractmethod
    def local_model_step(self, t, inputs):
        pass
    def run(self, duration):
        t = 0.0
        while t < duration:
            # receive subscribed data
            while not self.recv_queue.empty():
                attr_name, value = self.recv_queue.get()
                if attr_name in self.subscribe_handles:
                    self.subscribe_handles[attr_name] = value
            # call local model
            outputs = self.local_model_step(t, self.subscribe_handles)
            # send published data
            for name, val in outputs.items():
                self.send_queue.put((name, val))
                self.publish_handles[name] = val
            # time advance
            t += self.step
            # simulate RTI time regulation
            threading.Event().wait(self.step)
    def register_publish(self, name):
        self.publish_handles[name] = 0.0
    def register_subscribe(self, name):
        self.subscribe_handles[name] = 0.0

class GasTurbineFederate(HLAFederate):
    def __init__(self):
        super().__init__('GasTurbine', 0.001)
        self.register_publish('shaft_speed')
        self.register_publish('exhaust_temp')
        self.register_subscribe('fuel_flow')
        self.register_subscribe('compressor_inlet_pressure')
        self.J = 50.0   # inertia
        self.omega = 300.0
    def local_model_step(self, t, inputs):
        fuel = inputs.get('fuel_flow', 0.01)
        pin = inputs.get('compressor_inlet_pressure', 101325)
        torque_turbine = 500 * fuel * pin/101325
        torque_load = 200 * np.sqrt(self.omega/300)
        domega = (torque_turbine - torque_load)/self.J
        self.omega += domega * self.step
        temp = 1200 * fuel + 300
        return {'shaft_speed': self.omega, 'exhaust_temp': temp}

class COGAGFederate(HLAFederate):
    def __init__(self):
        super().__init__('COGAG', 0.001)
        self.register_publish('torque_output')
        self.register_subscribe('shaft_speed_gt1')
        self.register_subscribe('shaft_speed_gt2')
        self.torque_comb = 0.0
    def local_model_step(self, t, inputs):
        w1 = inputs.get('shaft_speed_gt1', 0)
        w2 = inputs.get('shaft_speed_gt2', 0)
        # gearbox combining logic
        if w1 > 100 and w2 > 100:
            combined = (w1 + w2)/2
        elif w1 > 100:
            combined = w1
        elif w2 > 100:
            combined = w2
        else:
            combined = 0
        self.torque_comb = 0.8 * combined
        return {'torque_output': self.torque_comb}

if __name__ == '__main__':
    gt = GasTurbineFederate()
    cogag = COGAGFederate()
    # simulate RTI federation
    thread1 = threading.Thread(target=gt.run, args=(5.0,))
    thread2 = threading.Thread(target=cogag.run, args=(5.0,))
    thread1.start(); thread2.start()
    thread1.join(); thread2.join()
    print('Final GT speed:', gt.omega, 'rpm')
    print('Final COGAG torque:', cogag.torque_comb, 'Nm')