Python Web 开发实战：基于 Flask + Vue 构建数字孪生平台

这里我们使用 Python 的 taos 客户端。创建超级表时带上标签，这样查询特定设备的数据会非常快。

# services/tdengine_client.py
import taos

conn = taos.connect(host="localhost", user="root", password="taosdata")

# 创建超级表（含标签）
conn.execute("""
CREATE TABLE IF NOT EXISTS devices (
    ts TIMESTAMP,
    temperature FLOAT,
    vibration FLOAT
) TAGS (
    device_id BINARY(32),
    type BINARY(16),
    location BINARY(64)
)
""")

def insert_sensor_data(device_id: str, data: dict):
    # 自动创建子表（如果不存在）
    conn.execute(f"""
    CREATE TABLE IF NOT EXISTS d_{device_id} USING devices
    TAGS ('{device_id}', '{data.get('type', 'unknown')}', '{data.get('location', '')}')
    """)
    # 插入数据
    sql = f"INSERT INTO d_{device_id} VALUES (NOW, {data['temperature']}, {data['vibration']})"
    conn.execute(sql)

实时状态推送（WebSocket）

为了保持前端视图的实时性，Flask SocketIO 是不错的选择。注意这里用 Redis 做缓存，避免每次更新都去查 TDengine。

# routes/websocket.py
from flask_socketio import SocketIO, emit

socketio = SocketIO(app, cors_allowed_origins="*")

@socketio.on('connect')
def handle_connect():
    emit('status', get_all_devices_latest_status())

def broadcast_device_update(device_id: str, status: dict):
    """由 MQTT 回调触发"""
    socketio.emit('device_update', {
        'device_id': device_id,
        'status': status
    })

3D 场景构建（Three.js + Vue）

加载 GLTF 工厂模型

在 Vue 中集成 Three.js，记得初始化场景、相机和渲染器。加载模型后，别忘了添加灯光，否则黑乎乎的一片。

<template>
  <div ref="sceneContainer"></div>
</template>

<script setup>
import * as THREE from 'three'
import { GLTFLoader } from 'three/examples/jsm/loaders/GLTFLoader'

let scene, camera, renderer, factoryModel

onMounted(async () => {
  // 初始化 Three.js
  scene = new THREE.Scene()
  camera = new THREE.PerspectiveCamera(75, window.innerWidth / window.innerHeight, 0.1, 1000)
  renderer = new THREE.WebGLRenderer({ antialias: true })
  renderer.setSize(window.innerWidth, window.innerHeight)
  sceneContainer.value.appendChild(renderer.domElement)

  // 加载工厂 GLTF 模型
  const loader = new GLTFLoader()
  const gltf = await loader.loadAsync('/models/factory.gltf')
  factoryModel = gltf.scene
  scene.add(factoryModel)

  // 添加灯光
  const light = new THREE.DirectionalLight(0xffffff, 1)
  light.position.set(5, 5, 5)
  scene.add(light)

  camera.position.z = 20
  animate()
})

function animate() {
  requestAnimationFrame(animate)
  renderer.render(scene, camera)
}
</script>

绑定设备状态到 3D 对象

拿到 WebSocket 推送的数据后，直接操作 3D 对象的材质颜色。比如温度过高变红，这里要注意遍历 Mesh 节点。

// 假设 GLTF 模型中设备命名为 "machine_01"
const machine = factoryModel.getObjectByName('machine_01')

// 监听 WebSocket 更新
socket.on('device_update', (data) => {
  if (data.device_id === 'machine_01') {
    // 温度 >80°C 时变红
    const color = data.status.temperature > 80 ? 0xff0000 : 0x00ff00
    machine.traverse((child) => {
      if (child.isMesh) child.material.color.setHex(color)
    })
  }
})

模型准备：使用 Blender 或 Siemens NX 导出带命名的 GLTF，确保名称与代码一致。

仿真预测 —— '如果...会怎样？'

故障传播规则（智慧工厂）

除了看当前状态，更重要的是推演后果。这里写一个简单的递归逻辑来模拟停机对下游的影响。

# services/failure_simulator.py

def simulate_machine_failure(machine_id: str):
    """模拟某设备停机对产线的影响"""
    affected = []
    # 规则 1: 下游设备因缺料停机
    downstream = get_downstream_machines(machine_id)
    for m in downstream:
        if not has_buffer_stock(m):
            affected.append(m)
            # 递归传播
            affected.extend(simulate_machine_failure(m))
    return affected

物理引擎集成（Cannon.js）

对于物体掉落、碰撞检测这类场景，物理引擎必不可少。注意同步 Three.js 的位置和 Cannon.js 的物理体。

import * as CANNON from 'cannon-es'

const world = new CANNON.World()
world.gravity.set(0, -9.82, 0)

// 为 3D 模型添加物理体
const boxBody = new CANNON.Body({ mass: 1 })
boxBody.addShape(new CANNON.Box(new CANNON.Vec3(1, 1, 1)))
world.addBody(boxBody)

// 同步 Three.js 与 Cannon.js
function updatePhysics() {
  world.step(1/60)
  boxMesh.position.copy(boxBody.position)
  boxMesh.quaternion.copy(boxBody.quaternion)
}

注意：复杂仿真建议在后端运行，前端仅可视化结果，避免浏览器卡顿。

场景实战

智慧工厂

• 数据源：PLC（Modbus）、振动传感器（MQTT）
• 孪生功能：
  • 实时显示设备温度、转速
  • 点击设备 → 查看历史趋势（ECharts）
  • '模拟停机'按钮 → 高亮受影响工位

智慧建筑

• BIM 模型：IFC → GLTF 转换
• 应急演练：
  • 触发'火灾' → 自动规划疏散路径（A* 算法）
  • AR 模式下，手机摄像头显示逃生箭头
• HVAC 仿真：
  • 根据室外温度、人流密度，动态调整空调功率
  • 可视化气流分布（粒子系统）

智慧城市

• 交通流仿真：
  • 基于 SUMO（开源交通模拟器）生成轨迹
  • 数字孪生平台接收车辆位置 → 更新 3D 道路状态
  • 优化信号灯配时：减少拥堵 15%

AR 增强现实（移动端）

Web AR 架构

利用 8th Wall 可以在浏览器里跑 SLAM，不需要装 App，这对现场巡检很友好。

[手机浏览器]
├── 8th Wall：摄像头 + SLAM 定位
├── A-Frame：声明式 3D 场景
└── Flask API：获取设备实时状态

AR 设备叠加

<!-- public/ar.html -->
<script src="https://aframe.io/releases/1.4.0/aframe.min.js"></script>
<script src="https://cdn.8thwall.com/web/xrweb/xrweb.js"></script>
<a-scene xrweb="appKey: YOUR_8THWALL_KEY">
  <!-- 动态加载设备标记 -->
  <a-entity gps-entity-place="latitude: 39.9042; longitude: 116.4074" scale="2 2 2">
    <a-box color="#FF0000" depth="1"></a-box>
    <a-text value="温度：{{temp}}°C" position="0 1.5 0"></a-text>
  </a-entity>
</a-scene>
<script>
  // 从 Flask 获取实时数据
  fetch('/api/device/machine_01')
    .then(r => r.json())
    .then(data => {
      document.querySelector('a-text').setAttribute('value', `温度：${data.temperature}°C`)
    })
</script>

优势：无需安装 App，扫码即用，适合现场巡检。

性能优化

大规模场景

• LOD（Level of Detail）：远距离使用低模
• 实例化渲染：相同设备（如路灯）批量绘制
• 数据采样：高频传感器数据降采样后推送

网络带宽

• 模型压缩：GLTF + Draco 压缩 → 体积减少 70%
• 增量更新：仅推送变化的设备状态

安全与权限

数据安全

• 设备认证：MQTT 使用 TLS + 客户端证书
• API 权限：RBAC 控制用户可操作的设备

仿真沙盒

• 禁止真实控制：仿真模式下所有'控制'仅影响虚拟模型
• 审计日志：记录所有仿真操作

未来方向

AI + 数字孪生

• 异常检测：LSTM 预测设备剩余寿命 → 在孪生体高亮
• 自动建模：无人机扫描 → 自动生成 GLTF

元宇宙集成

• 多用户协作：运维团队在同一个 3D 空间标注问题
• 数字资产交易：孪生模型作为 NFT 出售

总结

数字孪生的价值不在于'像'，而在于'用'——用虚拟世界优化物理世界。通过上述架构，我们可以实现从数据采集到仿真决策的完整闭环。