┌────────────┐  1. 边缘采集层：PLC/OPC-UA/Modbus → MQTT
│   物理产线 │
└────┬───────┘│实时数据（JSON over MQTT）
┌────▼───────┐  2. 孪生模型层：基于离散事件仿真（SimPy）
│   数字孪生 │
└────┬───────┘│REST/gRPC
┌────▼───────┐  3. 优化算法层：遗传算法、强化学习
│  优化引擎  │
└────┬───────┘│WebSocket
┌────▼───────┐  4. 可视化层：Dash/Three.js
│  3D 仪表盘 │
└────┬───────┘│OPC-UA Write
┌────▼───────┐  5. 闭环控制层：下发新排产到 PLC
│   PLC      │
└────────────┘

# opc2mqtt.py
import asyncio, json, os
from asyncua import Client
from paho.mqtt import publishPLC_URL = os.getenv("PLC_URL", "opc.tcp://192.168.0.10:4840")
MQTT_HOST = os.getenv("MQTT_HOST", "localhost")NODES = {"station1/CycleTime": "ns=2;s=Station1.CycleTime","station1/Status":    "ns=2;s=Station1.Status","station2/CycleTime": "ns=2;s=Station2.CycleTime","station2/Status":    "ns=2;s=Station2.Status","station3/CycleTime": "ns=2;s=Station3.CycleTime","station3/Status":    "ns=2;s=Station3.Status",
}async def bridge():async with Client(url=PLC_URL) as client:while True:payload = {}for topic, node_id in NODES.items():node = client.get_node(node_id)payload[topic] = await node.read_value()publish.single("factory/real", json.dumps(payload), hostname=MQTT_HOST)await asyncio.sleep(1)if __name__ == "__main__":asyncio.run(bridge())

# twin_model.py
import simpy, json, random, paho.mqtt.client as mqtt
from collections import dequeclass Station:def __init__(self, env, name, cycle_t, mtbf, repair_t):self.env = envself.name = nameself.cycle_t = cycle_t          # 标称节拍self.mtbf = mtbfself.repair_t = repair_tself.status = "RUN"self.queue = deque()self.env.process(self.work())self.env.process(self.failure())def work(self):while True:if self.queue:yield self.env.timeout(self.cycle_t)self.queue.popleft()else:yield self.env.timeout(1)def failure(self):while True:yield self.env.timeout(random.expovariate(1/self.mtbf))self.status = "DOWN"yield self.env.timeout(self.repair_t)self.status = "RUN"class TwinLine:def __init__(self):self.env = simpy.Environment()self.stations = [Station(self.env, "station1", cycle_t=15, mtbf=3600, repair_t=120),Station(self.env, "station2", cycle_t=20, mtbf=2880, repair_t=180),Station(self.env, "station3", cycle_t=10, mtbf=7200, repair_t=90),]self.env.process(self.source())self.mqtt_client = mqtt.Client()self.mqtt_client.on_message = self.on_mqttself.mqtt_client.connect("localhost")self.mqtt_client.subscribe("factory/real")self.mqtt_client.loop_start()def source(self):while True:self.stations[0].queue.append("job")yield self.env.timeout(12)  # 默认投料节拍def on_mqtt(self, client, userdata, msg):data = json.loads(msg.payload)for i in range(3):st = self.stations[i]key = f"station{i+1}/CycleTime"if key in data:st.cycle_t = data[key] * 0.001  # PLC 单位 msdef run(self):self.env.run(until=float('inf'))if __name__ == "__main__":TwinLine().run()

# optimizer.py
import random, json, requests, time
from deap import base, creator, toolsCACHED_API = "http://localhost:8080/simulate"  # 调用孪生仿真返回 makespandef eval_ind(ind):T, B = indresp = requests.post(CACHED_API, json={"T": T, "B": B}, timeout=10)return (resp.json()['makespan'],)creator.create("FitnessMin", base.Fitness, weights=(-1.0,))
creator.create("Individual", list, fitness=creator.FitnessMin)toolbox = base.Toolbox()
toolbox.register("attr_T", random.uniform, 10, 30)
toolbox.register("attr_B", random.randint, 1, 10)
toolbox.register("individual", tools.initCycle, creator.Individual,(toolbox.attr_T, toolbox.attr_B), n=1)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)toolbox.register("evaluate", eval_ind)
toolbox.register("mate", tools.cxBlend, alpha=0.3)
toolbox.register("mutate", tools.mutGaussian, mu=0, sigma=1, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)def main():pop = toolbox.population(n=30)for gen in range(10):offspring = algorithms.varAnd(pop, toolbox, cxpb=0.5, mutpb=0.2)fits = toolbox.map(toolbox.evaluate, offspring)for fit, ind in zip(fits, offspring):ind.fitness.values = fitpop = toolbox.select(offspring, k=len(pop))best = tools.selBest(pop, 1)[0]print(f"Gen {gen}: T={best[0]:.1f}, B={best[1]}, makespan={best.fitness.values[0]:.1f}")# 下发到 PLCrequests.post("http://localhost:8080/setT", json={"T": best[0]})return popif __name__ == "__main__":main()

# dashboard.py
import dash, json, paho.mqtt.client as mqtt
from dash import dcc, html, Input, Output
import plotly.graph_objects as goapp = dash.Dash(__name__)
app.layout = html.Div([dcc.Graph(id='live-oee'),dcc.Interval(id='timer', interval=1000)
])data = {"station1": {"cycle": 0, "status": "RUN"},"station2": {"cycle": 0, "status": "RUN"},"station3": {"cycle": 0, "status": "RUN"}}def on_msg(client, userdata, msg):global datadata = json.loads(msg.payload)client = mqtt.Client()
client.on_message = on_msg
client.connect("localhost")
client.subscribe("factory/real")
client.loop_start()@app.callback(Output('live-oee', 'figure'), Input('timer', 'n_intervals'))
def update(n):fig = go.Figure()fig.add_bar(x=list(data.keys()), y=[data[k].get("cycle", 0) for k in data])fig.update_layout(title="实时节拍 (ms)")return figif __name__ == "__main__":app.run_server(debug=True, port=8050)

# setT.py
from asyncua import Client
import asyncio, json, osPLC_URL = os.getenv("PLC_URL", "opc.tcp://192.168.0.10:4840")async def set_cycle_time(station_id, new_T_ms):async with Client(url=PLC_URL) as client:node = client.get_node(f"ns=2;s=Station{station_id}.TargetCycle")await node.write_value(float(new_T_ms))if __name__ == "__main__":import sysasyncio.run(set_cycle_time(int(sys.argv[1]), float(sys.argv[2])))