典型石化反應器案例架構：Physical AI 完整選型與 ROI 分析指南

台灣 31 年工業儀錶製造商 ATLANTIS 昶特出品｜針對化工廠工程師、設施管理者、採購決策者的 完整技術決策指南

前言：為什麼石化廠必須投資 Physical AI 監控系統

石化反應器不是簡單的容器。它是 溫度、壓力、流量、觸媒濃度 四大物理參數舞動的舞台。其中任何一個參數失控，可能導致：

• 爆炸與洩漏：反應釜內壓力瞬間飆升至 150 bar，造成容器脆裂
• 轉化率崩潰：溫度偏離 ±5°C，反應選擇性下跌 30~50%，產出廢品
• 產能癱瘓：風冷冷卻系統故障無人察覺，反應釜升溫至危險值，被迫全程停機
• 人身安全：值班人員疲勞監測，遺漏異常信號，成為決策延遲的元兇

過去 20 年，台灣石化業者多數依賴 人工巡檢 + 簡單指針儀錶 的被動監控模式。但在勞動力短缺、全球競爭加劇、ESG 法規收緊的今天，這套模式已不堪一擊。

Physical AI 改變了遊戲規則：透過分佈式感測器、邊緣計算、預測性維護演算法，工廠可以：

• 🎯 提前 2~4 小時偵測異常趨勢，在故障發生前採取行動
• 🎯 自動記錄全程數據，符合法規稽核與產品可追溯性要求
• 🎯 優化工藝參數，每個批次提升產率 3~8%
• 🎯 降低能耗，透過智慧冷卻系統節省 15~25% 蒸氣/冷卻水成本

本篇文章將帶你深入一座真實的北台灣石化廠案例，拆解其 Physical AI 監控系統的完整架構、投資成本、ROI 計算邏輯，以及選型時最容易踩的五大坑。最後會告訴你，ATLANTIS 昶特 31 年製造經驗中，如何幫助類似規模的廠商打造「一次投資、十年穩定」的工業級監控方案。

第一章：石化反應器的壓力溫度物理基礎

1.1 反應釜內發生了什麼？化學與物理的完美風暴

典型的石化反應流程如下：原料（例如乙烯）進入反應釜，在高溫高壓、觸媒催化下進行聚合反應，生成聚乙烯粒子。整個過程中，三個變數互相耦合：

三者互相制約：提高進料速率以增產，結果壓力升高；為了控制壓力而減少進料，產率又下來。工藝工程師的日常工作，就是在這三個參數之間找到最優平衡點。

1.2 現場案例：反應釜溫度失控的連鎖效應

這個案例是 2023 年真實發生在某個北台灣聚合廠。若早在 11:00 時就有 AI 系統自動偵測到「冷卻水流量異常」和「溫度上升趨勢」，主動發送 SMS 告警，操作員可以在 11:05 就開始應急程序，避免温度超過 220°C。三小時停機可以完全避免。

1.3 物理約束與法規要求

石化業的壓力溫度監測，並非單純的「預防故障」考量，還涉及法規合規：

第二章：典型石化反應器監控系統的硬體架構

2.1 測量點位規劃（Instrumentation Points）

一座「中等規模」的石化反應釜（單釜容積 1,000~2,000 公升），通常需要以下測量點位：

換句話說，最少需要 11 個獨立的感測器通道。若釜數較多（例如並聯 4 座反應釜進行多批次生產），總感測點數可達 44 個以上。

2.2 感測器選型與防爆等級

石化廠內若涉及易燃易爆氣體（例如乙烯、丙烯等），監測儀錶必須通過防爆認證。常見等級如下：

根據廠房位置與工藝風險評估（HAZOP 分析），採購部門會確定所需的防爆等級。這直接影響儀錶採購成本：

這正是為什麼投資 Physical AI 時，感測器選型必須與工藝風險等級掛勾。過度防爆認證會拉高成本；認證不足則違法且危險。

2.3 數據採集層：RS485 Modbus → LAN 網關 → AWS 架構

你的廠房採用典型的「現場層 → 邊緣層 → 雲端層」三層架構，每層通訊協議不同。感測器收集的數據必須經過多次轉換才能抵達雲端：

2.4 RS485 Modbus 通訊延遲與時序設計

RS485 Modbus 的通訊延遲比以太網高，但對於石化製程仍足以應對。完整的端到端延遲分解如下：

關鍵觀念：Physical AI 不是即時控制系統。反應釜的 PID 溫度控制（< 100ms）必須完全在本地邊緣執行（不依賴 RS485 或 AWS），而 Physical AI 的角色是「監測」與「預測」。若 RS485 因故障中斷，本地 PLC 仍能自主執行基本的溫度壓力控制，雲端 ML 功能暫停，待網路恢復後自動補回。

第三章：Physical AI 軟體層 — 異常檢測與預測模型

3.1 時間序列異常檢測模型

反應釜的溫度、壓力數據本質是「時間序列」— 連續的數據點，按時間順序排列。傳統的「固定閾值告警」（例如 T > 230°C 就告警）過於簡陋，容易誤報，也無法偵測漸進式故障（例如冷卻效率每天下降 1°C）。

物理 AI 系統採用的是「統計異常檢測」模型，核心思想是：學習歷史「正常」模式，當新來的數據偏離此模式時，判定為異常。

3.2 孤立森林（Isolation Forest）算法

孤立森林是一種無監督機器學習算法，特別適合工業應用，因為它：

• ✅ 無需標籤數據：不需要人工標註哪些歷史事件是「異常」
• ✅ 計算效率高：可在邊緣節點（如樹莓派）上即時執行
• ✅ 自動適應新環境：當廠房轉換新產品配方時，自動重新學習「正常」模式
• ✅ 解釋性好：告訴你「異常分數」，工程師易於理解

# 孤立森林異常檢測的偽代碼（Python-like）

import pandas as pd
from sklearn.ensemble import IsolationForest
import numpy as np

# 1. 準備歷史數據（過去 30 天的正常運作記錄）
historical_data = pd.read_csv('normal_operation_30days.csv')
# 格式: [時間戳, 釜內溫度, 釜內壓力, 進液溫度, 冷卻流量, ...]

# 2. 訓練孤立森林模型
iso_forest = IsolationForest(
    contamination=0.05,  # 假設 5% 的數據為異常
    random_state=42,
    n_estimators=100
)
iso_forest.fit(historical_data[['temperature', 'pressure', 'coolant_flow']])

# 3. 對新來的實時數據進行異常分數計算
new_sample = {
    'temperature': 232.5,  # °C
    'pressure': 110,       # bar
    'coolant_flow': 12.3   # L/min
}

anomaly_score = iso_forest.decision_function([new_sample.values()])
is_anomaly = iso_forest.predict([new_sample.values()])  # 返回 -1 (異常) 或 1 (正常)

if is_anomaly[0] == -1:
    print(f"⚠️ 檢測到異常，異常分數：{anomaly_score[0]:.3f}")
    send_alert_sms("溫度升高至 232.5°C，且壓力異常高於預期")
else:
    print(f"✓ 數據正常，異常分數：{anomaly_score[0]:.3f}")

3.3 長短期記憶網絡（LSTM）— 預測故障前驅信號

如果說孤立森林是「檢測當下異常」，那麼 LSTM 則是「預測未來異常」。LSTM 是深度學習中最常見的時間序列模型，能夠學習長期依賴關係，進而預測未來 N 小時內溫度、壓力的演變趨勢。

實務案例：

# LSTM 預測模型示意（TensorFlow / Keras）

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
import numpy as np

# 1. 準備訓練數據
# X_train: (樣本數, 時間步長, 特徵數)
# 例如 (5000, 24, 5) = 5000 個批次，每個批次過去 24 小時，5 個特徵（T, P, Q_cool, T_diff, 等）
# y_train: (樣本數, 預測步長) 
# 例如 (5000, 4) = 預測未來 4 小時的溫度

# 2. 構建 LSTM 模型
model = Sequential([
    LSTM(64, activation='relu', input_shape=(24, 5), return_sequences=True),
    Dropout(0.2),
    LSTM(32, activation='relu'),
    Dropout(0.2),
    Dense(16, activation='relu'),
    Dense(4)  # 輸出層：預測未來 4 小時
])

model.compile(optimizer='adam', loss='mse', metrics=['mae'])

# 3. 訓練模型
model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.2)

# 4. 實時預測
current_24h_data = fetch_latest_24h_sensor_data()  # 形狀 (1, 24, 5)
forecast_4h = model.predict(current_24h_data)  # 預測未來 4 小時溫度

# 5. 如果預測顯示溫度會升高超過 230°C，提前告警
if forecast_4h[0].max() > 230:
    print("⚠️ 預測：未來 4 小時內溫度將超過 230°C，建議提前調整冷卻流量")

3.4 物理約束神經網絡（PINN）— 融合領域知識

傳統的 LSTM 是「純數據驅動」— 完全從歷史數據學習模式。但石化反應其實受物理定律制約（質量平衡、能量平衡等），若能將這些約束編碼進神經網絡，模型會更精準且泛化能力更強。

PINN 的核心是在損失函數中融合「物理方程殘差」：

PINN 的優勢在於 小樣本學習：若一家新廠房的歷史數據只有 10 天（而非通常的 90 天），PINN 仍能透過物理約束進行有效訓練，而純 LSTM 則會嚴重過擬合。

第四章：北台灣石化廠案例研究 — 完整架構與成本分析

4.1 案例背景：某聚合廠的監控升級旅程

北台灣某聚合廠（匿名），成立於 2005 年，專業生產聚乙烯（PE）與聚丙烯（PP）顆粒。廠房運營規模：

4.2 問題診斷：為什麼停機頻繁？

廠方進行了一次詳細的故障根本原因分析（RCA），發現：

總月度成本影響：顯著；年度成本：相當可觀

4.3 解決方案設計：分階段導入 Physical AI

廠方與 ATLANTIS 昶特團隊合作，制定了一份「12 個月分階段升級計畫」：

第一階段（1~2 月）：RS485 Modbus 感測器與網關部署

此階段完成後，廠方掌握了「完整的實時數據可見性」— 4 座釜的溫度、壓力、流量數據每 10 秒同步一次至本地 PLC 緩衝區。

第二階段（3~5 月）：邊緣計算節點與異常檢測

此階段完成後，廠方擁有「異常自動檢測 + 主動告警」能力。例如：

• 冷卻水流量突降 > 20% → 立即 SMS 告警操作員，他可在 2 分鐘內巡檢並清洗堵塞
• 反應釜溫度連續上升，預測 2 小時後超過安全值 → 提前通知主管，決策是否降低進料速率

第三階段（6~9 月）：雲端數據庫 + 長期趨勢分析（客戶自行承擔）

雲端部分完成後，廠方能夠：

• 查閱任意時間段的完整歷史記錄（符合法規稽核需求）
• 自動計算「本月平均轉化率」、「產品等級分佈」等 KPI
• 預測「下週哪台釜需進行預防性維護」，提前排程，避免緊急停機

第四階段（10~12 月）：優化與穩定化

4.4 投資成本總結（RS485 Modbus 方案）

4.5 成效評估：6 個月後的數據

系統上線 6 個月後，廠方進行成效評估：

4.6 ROI 與回本週期計算

回本週期計算

總投資 / 年度淨節省 = 報價制 / 報價制 ≈ 1.23 年

考慮到維保成本 報價制/年，實際回本週期約 1.5 年。第二年開始，廠方每年淨獲益約 報價制，持續 10 年預期累計收益將達 報價制。

換個角度看：

第五章：選型決策框架 — 「我的廠房適合導入嗎？」

5.1 快速自檢清單

並非所有廠房都適合投資 Physical AI。以下清單幫你判斷：

若上表中大部分指標都落在「✅ 適合導入」，那麼 Physical AI 投資會有明顯 ROI。若多數在「⚠️」，需要更細緻的成本效益分析。若多數在「❌」，暫時考慮投資簡單的感測器與本地 PLC 控制即可。

5.2 五大常見踩坑

坑 #1：買了感測器但沒有配套的數據採集系統

現象：廠方購置了 ATLANTIS 的高精度壓力計，但卻用老舊的類比式儀錶板（指針），無法實現自動記錄或告警。

後果：感測器的價值無法發揮，與購置便宜儀錶的效果沒什麼差別。

預防：感測器採購必須與資料採集系統綁定採購。完整的方案應包括：感測器 + DAQ 模組 + PLC + 軟體。ATLANTIS 提供完整的整合諮詢服務，可幫助設計端到端的系統架構。

坑 #2：過度依賴進口品牌，忽視本土解決方案

現象：廠方迷信進口品牌（如 WIKA、Yokogawa 等），全套系統採購成本高達 報價制，而相同功能的本土方案只需 報價制。

後果：投資成本高，ROI 週期拉長至 2~3 年，削弱決策層的支持。

預防：在功能相同的前提下，優先選擇符合防爆等級、精度等級、且支援本地快速維修的方案。ATLANTIS 31 年本土製造經驗，「整合供應鏈在台灣」意味著：備品更快、維修成本更低、技術人員更容易培訓。

坑 #3：忽視操作人員的培訓與變革管理

現象：系統上線後，操作人員因為不熟悉新界面，反而增加了誤操作。或是，告警規則設定不當，導致頻繁誤報，最終被人為關閉。

後果：投資無法轉化為實際效益。

預防：系統導入成本中，應分配 15~20% 用於人員培訓與變革管理。包括：操作人員的界面訓練、維護人員的故障診斷培訓、管理層的 KPI 監控培訓。ATLANTIS 在部署時會提供 2 週的現場駐廠培訓。

坑 #4：告警規則過於敏感，導致「告警疲勞」

現象：系統每天發送 50+ 條告警，值班人員無法區分優先級，最終所有告警都被忽視。

後果：真正的異常信號被淹沒在噪音中。

預防：告警規則應分層設計：(1) 致命異常（P > 150 bar）→ 自動關閉進料、啟動應急冷卻；(2) 重度異常（T > 230°C）→ SMS 告警主管；(3) 輕度異常（溫度上升趨勢）→ 日報形式通知。首 2 週是「標定期」，根據實際運營數據不斷調整閾值。

坑 #5：沒有設計「降級模式」（Fallback Mode）

現象：邊緣計算節點或雲端服務故障，整個監控系統癱瘓，廠房無法決策是否繼續運作。

後果：引發人為故障或安全隱患。

預防：系統設計應包含「本地 PLC 獨立運作模式」：即使雲端無法連線，本地 PLC 仍能執行基本的溫度控制與壓力監測。智慧化功能（如預測性維護）在雲端恢復後再補回。這通常需要額外 10~15% 的軟體開發成本。

第六章：ATLANTIS 昶特的完整解決方案

6.1 感測器產品線

ATLANTIS 針對石化反應器應用，提供以下核心產品：

壓力量測系列

圖說：ATLANTIS 隔膜式壓力計 — Zone 1 防爆認證，適用於乙烯、丙烯等易燃氣體環境

隔膜式壓力計

• ✅ 防爆認證：Zone 0 / Zone 1 可選
• ✅ 精度：0.6 級（≤ ±0.6% FS，符合 ASME 標準）
• ✅ 隔膜材質：316 不銹鋼或鈦合金（抗腐蝕）
• ✅ 輸出選項：指針、4~20 mA、Modbus RTU
• ✅ 溫度補償：內建微調機制，-20~80°C 工作範圍

典型應用：反應釜工作壓力、安全閥進出壓力、夾套冷卻壓力監測。

溫度量測系列

圖說：ATLANTIS Pt100 RTD A 級精度溫度傳感器 — 廣泛用於反應釜溫度監測

Pt100 RTD（電阻式溫度計）

• ✅ 精度等級：A 級（±0.15 + 0.002|T|）或 B 級（±0.3 + 0.005|T|）
• ✅ 防爆套件：可選，符合 Zone 1 要求
• ✅ 反應時間：< 10 秒（浸沒型）
• ✅ 輸出形式：2/3/4 線制，直接相容標準 PLC 模組
• ✅ 工作溫度：-200~+600°C（等級差異）

K 型熱電偶（用於高溫、快速響應場景）

• ✅ 精度：±2.2°C（或精度等級 ±0.75%）
• ✅ 反應時間：1~2 秒（最快反應）
• ✅ 工作溫度：-200~+1372°C
• ✅ 成本：比 Pt100 便宜 30%

典型應用：Pt100 用於釜內多點溫度監測（精度要求高）；K 型熱電偶用於冷卻系統進出口（快速響應需求）。

流量量測系列

圖說：ATLANTIS 渦輪流量計 — 適用於冷卻液與進料流量監測

渦輪流量計

• ✅ 精度：±2% 讀值（符合工業標準）
• ✅ 防爆認證：可選
• ✅ 訊號輸出：脈衝 + 4~20 mA 雙模式
• ✅ 量程比：1:10~1:15（覆蓋寬流量範圍）
• ✅ 絕對精度：特別適合冷卻水與液體進料測量

科氏質量流量計（質量式）（用於高精度計費或成分追蹤）

• ✅ 精度：±0.2% 讀值（最高精度）
• ✅ 直接測質量，無需溫度補償
• ✅ 抗振性強（反應釜振動環境友好）
• ✅ 成本：比渦輪流量計高 2~3 倍，但適合關鍵物料計量

6.2 系統集成方案

ATLANTIS 不僅提供單一儀錶，更提供完整的「感測器 + PLC + 軟體 + 維保」一體化方案。

6.3 ATLANTIS 的技術優勢

第七章：20 大常見問題 FAQ（Physical AI 石化應用）

第八章：內連延伸閱讀

第九章：結論與行動呼籲

核心要點回顧

投資決策檢查清單

若你的廠房符合以下條件，現在正是投資 Physical AI 的最佳時機：

• ✅ 年營收 > 5,000 萬，或停機成本 > 報價制/小時
• ✅ 反應釜數量 ≥ 2 座
• ✅ 現有故障成本 報價制/年
• ✅ 已有基本的自動化基礎（PLC 或簡易 DCS）
• ✅ 願意分配預算進行員工培訓與變革管理
• ✅ 有明確的 KPI（如產率、品質、能耗）改善目標

若上述大部分都符合，立即與 ATLANTIS 昶特團隊聯絡進行「免費現場 RS485 Modbus 評估」。

與 ATLANTIS 的下一步

第十章：技術附錄 — RS485 Modbus 到 AWS 的完整部署

10.0 架構示意圖

10.1 RS485 Modbus 感測器讀取 + AWS 上傳（完整工作流）

# ===== 檔案名稱：modbus_edge_gateway.py =====
# ===== 執行環境：Raspberry Pi 4 / 邊緣計算盒，Python 3.8+ =====
# ===== 依賴庫：pip install pymodbus paho-mqtt numpy scikit-learn =====

from pymodbus.client import ModbusSerialClient as ModbusClient
from pymodbus.exceptions import ModbusException
import paho.mqtt.client as mqtt
import json
import time
import logging
from datetime import datetime
from sklearn.ensemble import IsolationForest
import joblib
import struct

# ===== 日誌配置 =====
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# ===== 配置區塊 =====
MODBUS_PORT = '/dev/ttyUSB0'  # 樹莓派 USB 轉 RS485 的設備路徑
MODBUS_BAUDRATE = 9600        # RS485 鮑率
MODBUS_TIMEOUT = 1            # 讀取超時（秒）

MQTT_BROKER = "192.168.1.100"  # 本地 MQTT broker
MQTT_PORT = 1883
MQTT_TOPIC_DATA = "factory/reactor/sensor_data"    # 發佈感測器數據
MQTT_TOPIC_ALERT = "factory/reactor/alerts"        # 發佈告警
MQTT_TOPIC_COMMANDS = "factory/reactor/commands"   # 訂閱控制指令

AWS_IOT_ENDPOINT = "your-iot-endpoint.iot.ap-northeast-1.amazonaws.com"
AWS_IOT_PORT = 8883
AWS_IOT_TOPIC = "factory/reactor/aws_sync"

# ===== Modbus 感測器地址對應表 =====
MODBUS_MAP = {
    # 釜 1 - 3 支溫度計
    'reactor_1_temp_bottom': {'address': 1, 'register': 0, 'scale': 0.1},
    'reactor_1_temp_middle': {'address': 1, 'register': 1, 'scale': 0.1},
    'reactor_1_temp_top': {'address': 1, 'register': 2, 'scale': 0.1},
    'reactor_1_pressure': {'address': 2, 'register': 0, 'scale': 0.1},
    
    # 釜 2~4（簡化表示）
    'reactor_2_temp': {'address': 3, 'register': 0, 'scale': 0.1},
    'reactor_2_pressure': {'address': 4, 'register': 0, 'scale': 0.1},
    
    # 冷卻系統
    'coolant_inlet_temp': {'address': 17, 'register': 0, 'scale': 0.1},
    'coolant_outlet_temp': {'address': 18, 'register': 0, 'scale': 0.1},
    'coolant_flow': {'address': 19, 'register': 0, 'scale': 0.1},
    'coolant_pressure_in': {'address': 20, 'register': 0, 'scale': 0.1},
}

# ===== 孤立森林模型載入 =====
try:
    iso_forest = joblib.load('./models/isolation_forest.pkl')
    scaler = joblib.load('./models/scaler.pkl')
    logger.info("✓ 孤立森林模型已載入")
except FileNotFoundError:
    logger.warning("⚠ 異常檢測模型未找到，將使用簡單固定閾值")
    iso_forest = None

# ===== Modbus 客戶端初始化 =====
def init_modbus_client():
    """
    初始化 RS485 Modbus RTU 客戶端
    """
    client = ModbusClient(
        method='rtu',
        port=MODBUS_PORT,
        baudrate=MODBUS_BAUDRATE,
        timeout=MODBUS_TIMEOUT,
        stopbits=1,
        bytesize=8,
        parity='N'
    )
    
    if not client.connect():
        logger.error("❌ Modbus 連接失敗，退出程式")
        exit(1)
    
    logger.info(f"✓ Modbus 連接成功 ({MODBUS_PORT}, {MODBUS_BAUDRATE} bps)")
    return client

# ===== 讀取單一 Modbus 寄存器 =====
def read_modbus_register(client, address, register_offset, scale=1.0):
    """
    讀取 Modbus 保持寄存器（地址, 偏移量, 縮放因數）
    Modbus 的保持寄存器是 16-bit 有號數，範圍 -32768 ~ 32767
    """
    try:
        # 讀取保持寄存器 (FC 03)
        result = client.read_holding_registers(
            address=register_offset,
            count=1,
            slave=address
        )
        
        if result.isError():
            logger.warning(f"Modbus 讀取失敗 (地址 {address}, 寄存器 {register_offset})")
            return None
        
        raw_value = result.registers[0]
        # 若感測器傳回的是帶符號整數，進行轉換
        if raw_value > 32767:
            raw_value = raw_value - 65536
        
        scaled_value = raw_value * scale
        return scaled_value
        
    except ModbusException as e:
        logger.error(f"Modbus 例外：{e}")
        return None

# ===== 輪詢所有感測器 =====
def poll_all_sensors(client):
    """
    循環讀取所有 Modbus 感測器，返回字典
    """
    sensor_data = {}
    
    for sensor_name, modbus_info in MODBUS_MAP.items():
        value = read_modbus_register(
            client,
            modbus_info['address'],
            modbus_info['register'],
            modbus_info['scale']
        )
        
        if value is not None:
            sensor_data[sensor_name] = value
        else:
            sensor_data[sensor_name] = None
            logger.warning(f"⚠ {sensor_name} 讀取失敗")
    
    return sensor_data

# ===== 異常檢測 =====
def detect_anomaly(sensor_data):
    """
    使用孤立森林進行異常檢測
    """
    if iso_forest is None:
        return None, 0
    
    # 提取關鍵特徵
    features = [
        sensor_data.get('reactor_1_temp_middle', 0),
        sensor_data.get('reactor_1_pressure', 0),
        sensor_data.get('coolant_flow', 0),
        sensor_data.get('coolant_inlet_temp', 0),
        sensor_data.get('coolant_outlet_temp', 0),
    ]
    
    # 移除 None 值
    features = [f if f is not None else 0 for f in features]
    
    # 標準化
    features_scaled = scaler.transform([features])
    
    # 異常檢測
    anomaly_score = iso_forest.decision_function(features_scaled)[0]
    is_anomaly = iso_forest.predict(features_scaled)[0] == -1
    
    return is_anomaly, anomaly_score

# ===== MQTT 連接回調 =====
def on_mqtt_connect(client, userdata, flags, rc):
    if rc == 0:
        logger.info("✓ MQTT 連接成功")
        client.subscribe(MQTT_TOPIC_COMMANDS)
    else:
        logger.error(f"❌ MQTT 連接失敗，代碼 {rc}")

def on_mqtt_message(client, userdata, msg):
    """
    接收來自雲端或 PLC 的控制指令
    """
    try:
        payload = json.loads(msg.payload.decode())
        logger.info(f"收到指令：{payload}")
        # 根據需求執行本地動作（例如調整 PLC 參數）
    except json.JSONDecodeError:
        logger.warning(f"MQTT 訊息格式錯誤：{msg.payload}")

# ===== 主程式迴圈 =====
def main():
    """
    邊緣網關的主工作流程
    """
    logger.info("="*60)
    logger.info("石化反應器 RS485 Modbus 邊緣網關啟動")
    logger.info("="*60)
    
    # 1. 初始化 Modbus
    modbus_client = init_modbus_client()
    
    # 2. 初始化 MQTT
    mqtt_client = mqtt.Client()
    mqtt_client.on_connect = on_mqtt_connect
    mqtt_client.on_message = on_mqtt_message
    
    logger.info(f"連接 MQTT Broker：{MQTT_BROKER}:{MQTT_PORT}")
    mqtt_client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
    mqtt_client.loop_start()
    
    # 3. 主循環：每 10 秒讀取一次所有感測器
    poll_count = 0
    while True:
        try:
            logger.info(f"\n[輪詢 #{poll_count}] 開始讀取 RS485 感測器...")
            start_time = time.time()
            
            # 讀取所有感測器
            sensor_data = poll_all_sensors(modbus_client)
            
            # 異常檢測
            is_anomaly, anomaly_score = detect_anomaly(sensor_data)
            
            # 構造 JSON 訊息
            message = {
                'timestamp': datetime.now().isoformat(),
                'sensors': sensor_data,
                'anomaly_detected': is_anomaly,
                'anomaly_score': float(anomaly_score) if anomaly_score is not None else None,
                'poll_count': poll_count
            }
            
            # 發佈至本地 MQTT（稍後由另一個進程轉發至 AWS）
            mqtt_client.publish(
                MQTT_TOPIC_DATA,
                json.dumps(message, default=str)
            )
            
            elapsed = time.time() - start_time
            logger.info(f"✓ 讀取完成，耗時 {elapsed:.2f} 秒")
            
            if is_anomaly:
                logger.warning(f"⚠ 檢測到異常，分數：{anomaly_score:.3f}")
                mqtt_client.publish(MQTT_TOPIC_ALERT, json.dumps(message, default=str))
            
            poll_count += 1
            
            # 等待至 10 秒才進行下一次讀取
            time.sleep(max(0, 10 - elapsed))
            
        except KeyboardInterrupt:
            logger.info("\n程式被用戶中斷")
            break
        except Exception as e:
            logger.error(f"❌ 主迴圈異常：{e}")
            time.sleep(5)
    
    # 清理
    modbus_client.close()
    mqtt_client.loop_stop()
    logger.info("邊緣網關已停止")

if __name__ == '__main__':
    main()

10.2 邊緣節點 MQTT 到 AWS IoT Core 的數據轉發

# ===== 檔案名稱：mqtt_to_aws_bridge.py =====
# ===== 功能：本地 MQTT broker 的數據轉發至 AWS IoT Core =====

import paho.mqtt.client as mqtt
import json
import ssl
import logging
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ===== AWS IoT 配置 =====
AWS_IOT_ENDPOINT = "your-iot-endpoint.iot.ap-northeast-1.amazonaws.com"
AWS_IOT_PORT = 8883
AWS_IOT_CLIENT_ID = "EdgeGateway_Reactor_01"

# 憑證路徑（需從 AWS IoT 主控台下載）
AWS_CERT_PATH = "/home/pi/certs/certificate.pem.crt"
AWS_KEY_PATH = "/home/pi/certs/private.key"
AWS_ROOT_CA_PATH = "/home/pi/certs/AmazonRootCA1.pem"

LOCAL_MQTT_BROKER = "192.168.1.100"
LOCAL_MQTT_PORT = 1883
LOCAL_MQTT_SUBSCRIBE_TOPIC = "factory/reactor/sensor_data"
AWS_IOT_PUBLISH_TOPIC = "factory/reactor/aws_sync"

# ===== AWS IoT MQTT 客戶端初始化 =====
def init_aws_iot_client():
    """
    連接至 AWS IoT Core
    """
    aws_client = AWSIoTMQTTClient(AWS_IOT_CLIENT_ID)
    aws_client.configureEndpoint(AWS_IOT_ENDPOINT, AWS_IOT_PORT)
    aws_client.configureCredentials(AWS_ROOT_CA_PATH, AWS_KEY_PATH, AWS_CERT_PATH)
    
    # 配置 MQTT 連接參數
    aws_client.configureAutoReconnectBackoffTime(1, 32, 20)
    aws_client.configureOfflinePublishQueueing(-1)  # 無限隊列
    aws_client.configureDrainingFrequency(2)
    aws_client.configureConnectDisconnectTimeout(10)
    aws_client.configureMQTTOperationTimeout(5)
    
    try:
        aws_client.connect()
        logger.info("✓ AWS IoT Core 連接成功")
        return aws_client
    except Exception as e:
        logger.error(f"❌ AWS IoT 連接失敗：{e}")
        return None

# ===== 本地 MQTT 訊息接收（回調） =====
def on_local_message(client, userdata, msg):
    """
    接收本地 MQTT 訊息，轉發至 AWS
    """
    try:
        payload = json.loads(msg.payload.decode())
        
        # 添加邊緣節點簽名
        payload['edge_node'] = 'EdgeGateway_Reactor_01'
        payload['received_at_edge'] = time.time()
        
        # 發佈至 AWS
        aws_client = userdata['aws_client']
        if aws_client:
            aws_client.publish(
                AWS_IOT_PUBLISH_TOPIC,
                json.dumps(payload),
                1  # QoS = 1（至少一次傳輸）
            )
            logger.info(f"✓ 轉發至 AWS：{payload['timestamp']}")
        else:
            logger.warning("AWS 客戶端未就緒，暫存訊息至本地隊列")
            
    except json.JSONDecodeError:
        logger.error(f"JSON 解析失敗：{msg.payload}")
    except Exception as e:
        logger.error(f"轉發失敗：{e}")

# ===== 主程式 =====
def main():
    logger.info("="*60)
    logger.info("MQTT 到 AWS IoT Core 數據橋接啟動")
    logger.info("="*60)
    
    # 1. 初始化 AWS IoT 連接
    aws_client = init_aws_iot_client()
    if not aws_client:
        logger.error("無法連接 AWS IoT，程式退出")
        exit(1)
    
    # 2. 初始化本地 MQTT 客戶端（作為訂閱者）
    local_mqtt = mqtt.Client()
    local_mqtt.user_data_set({'aws_client': aws_client})
    local_mqtt.on_message = on_local_message
    
    logger.info(f"連接本地 MQTT Broker：{LOCAL_MQTT_BROKER}:{LOCAL_MQTT_PORT}")
    local_mqtt.connect(LOCAL_MQTT_BROKER, LOCAL_MQTT_PORT, keepalive=60)
    local_mqtt.subscribe(LOCAL_MQTT_SUBSCRIBE_TOPIC)
    local_mqtt.loop_forever()

if __name__ == '__main__':
    main()

10.1 孤立森林模型的完整訓練與部署示範

# ===== 檔案名稱：train_isolation_forest.py =====
# ===== 環境：Python 3.8+，scikit-learn 0.24+ =====

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import json
from datetime import datetime

# ===== 步驟 1：載入 30 天的歷史正常運作數據 =====
def load_historical_data(csv_file):
    """
    CSV 格式：
    timestamp, temperature, pressure, coolant_flow, coolant_inlet_temp, coolant_outlet_temp
    2024-01-01 08:00:00, 195.5, 75.2, 12.5, 28.0, 35.5
    ...
    """
    df = pd.read_csv(csv_file)
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    return df

# ===== 步驟 2：特徵工程（從原始信號提取有意義的特徵）=====
def extract_features(df):
    """
    提取用於異常檢測的特徵
    """
    features = df[['temperature', 'pressure', 'coolant_flow']].copy()
    
    # 新增衍生特徵：溫度差、流量與壓力的交互項
    features['temp_coolant_delta'] = (
        df['temperature'] - (df['coolant_inlet_temp'] + df['coolant_outlet_temp']) / 2
    )
    features['flow_pressure_ratio'] = df['coolant_flow'] / (df['pressure'] + 0.1)  # 避免除以 0
    
    return features

# ===== 步驟 3：數據預處理 =====
def preprocess_data(features):
    """
    標準化數據，使各特徵均值為 0、標準差為 1
    """
    scaler = StandardScaler()
    scaled_features = scaler.fit_transform(features)
    return scaled_features, scaler

# ===== 步驟 4：訓練孤立森林 =====
def train_isolation_forest(scaled_features, contamination=0.05):
    """
    contamination: 預期異常數據的百分比（通常 1~10%）
    """
    iso_forest = IsolationForest(
        n_estimators=100,
        contamination=contamination,
        random_state=42,
        n_jobs=-1  # 使用全部 CPU 核心加速
    )
    iso_forest.fit(scaled_features)
    return iso_forest

# ===== 步驟 5：模型驗證與性能評估 =====
def evaluate_model(iso_forest, scaled_features, features_df):
    """
    計算異常分數分佈，幫助確定閾值
    """
    anomaly_scores = iso_forest.decision_function(scaled_features)
    predictions = iso_forest.predict(scaled_features)
    
    print(f"異常分數統計:")
    print(f"  最小值：{anomaly_scores.min():.4f}")
    print(f"  最大值：{anomaly_scores.max():.4f}")
    print(f"  平均值：{anomaly_scores.mean():.4f}")
    print(f"  標準差：{anomaly_scores.std():.4f}")
    print(f"  檢測到的異常樣本數：{(predictions == -1).sum()}")
    
    # 保存異常樣本供人工審視
    anomaly_mask = predictions == -1
    anomalies = features_df[anomaly_mask]
    anomalies.to_csv('detected_anomalies.csv', index=False)
    print(f"已保存異常樣本至 detected_anomalies.csv")

# ===== 步驟 6：模型序列化與部署 =====
def save_model(iso_forest, scaler, output_dir='./models'):
    """
    將模型與 Scaler 保存為 pickle，方便後續載入使用
    """
    import os
    os.makedirs(output_dir, exist_ok=True)
    
    joblib.dump(iso_forest, f'{output_dir}/isolation_forest.pkl')
    joblib.dump(scaler, f'{output_dir}/scaler.pkl')
    print(f"模型已保存至 {output_dir}/")

# ===== 步驟 7：實時異常檢測函式 =====
def detect_anomaly_realtime(
    temperature,
    pressure,
    coolant_flow,
    coolant_inlet_temp,
    coolant_outlet_temp,
    iso_forest,
    scaler,
    anomaly_threshold=-0.5
):
    """
    對新傳入的實時數據進行異常判定
    anomaly_threshold: 異常分數閾值，越小越敏感
    """
    # 構造特徵
    features = np.array([
        temperature,
        pressure,
        coolant_flow,
        temperature - (coolant_inlet_temp + coolant_outlet_temp) / 2,
        coolant_flow / (pressure + 0.1)
    ]).reshape(1, -1)
    
    # 標準化
    scaled_features = scaler.transform(features)
    
    # 計算異常分數
    anomaly_score = iso_forest.decision_function(scaled_features)[0]
    is_anomaly = iso_forest.predict(scaled_features)[0] == -1
    
    return {
        'is_anomaly': is_anomaly,
        'anomaly_score': anomaly_score,
        'confidence': 1 - (anomaly_score / (-0.5))  # 簡化的信心度計算
    }

# ===== 主程式執行流程 =====
if __name__ == '__main__':
    print("=" * 60)
    print("石化反應器 - 孤立森林異常檢測模型訓練")
    print("=" * 60)
    
    # 1. 載入數據
    print("\n[1/6] 載入歷史數據...")
    df = load_historical_data('normal_operation_30days.csv')
    print(f"  載入 {len(df)} 筆記錄，時間跨度：{df['timestamp'].min()} 至 {df['timestamp'].max()}")
    
    # 2. 特徵工程
    print("\n[2/6] 執行特徵工程...")
    features = extract_features(df)
    print(f"  生成特徵數量：{features.shape[1]}")
    
    # 3. 數據預處理
    print("\n[3/6] 標準化數據...")
    scaled_features, scaler = preprocess_data(features)
    print(f"  標準化完成，形狀：{scaled_features.shape}")
    
    # 4. 訓練模型
    print("\n[4/6] 訓練孤立森林（100 棵決策樹）...")
    iso_forest = train_isolation_forest(scaled_features, contamination=0.05)
    print("  訓練完成")
    
    # 5. 模型評估
    print("\n[5/6] 模型性能評估...")
    evaluate_model(iso_forest, scaled_features, features)
    
    # 6. 保存模型
    print("\n[6/6] 保存模型...")
    save_model(iso_forest, scaler)
    
    print("\n" + "=" * 60)
    print("訓練流程完成！模型已可用於實時異常檢測")
    print("=" * 60)

10.2 邊緣計算節點的 MQTT 告警發送邏輯

# ===== 檔案名稱：edge_node_mqtt_alerts.py =====
# ===== 執行環境：Siemens MindConnect IoT Gateway 或 Raspberry Pi 4 =====

import paho.mqtt.client as mqtt
import json
import time
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
import joblib

# ===== 全局配置 =====
MQTT_BROKER = "192.168.1.100"  # 本地 PLC 或 MQTT Broker 地址
MQTT_PORT = 1883
MQTT_SUBSCRIBE_TOPIC = "factory/reactor/data"  # 訂閱主題（來自 PLC）
MQTT_PUBLISH_TOPIC = "factory/alerts"  # 發佈告警的主題

# 載入預訓練的孤立森林模型
iso_forest = joblib.load('./models/isolation_forest.pkl')
scaler = joblib.load('./models/scaler.pkl')

# SMS / Email 配置
ALERT_RECIPIENTS = {
    'sms': ['0912345678', '0987654321'],  # 操作員手機
    'email': ['maintenance@factory.com', 'manager@factory.com']
}

# 告警規則設定
ALERT_LEVELS = {
    'CRITICAL': {'temp_max': 240, 'pressure_max': 140, 'requires': 'auto_mitigation'},
    'HIGH': {'temp_max': 230, 'pressure_max': 120, 'requires': 'sms_notification'},
    'MEDIUM': {'temp_max': 220, 'requires': 'email_notification'},
}

# ===== MQTT 連接回調 =====
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("[MQTT] 連接成功，訂閱主題...")
        client.subscribe(MQTT_SUBSCRIBE_TOPIC)
    else:
        print(f"[MQTT] 連接失敗，代碼 {rc}")

# ===== 接收 PLC 發送的數據 =====
def on_message(client, userdata, msg):
    """
    接收 JSON 格式的感測器數據
    """
    try:
        payload = json.loads(msg.payload.decode())
        timestamp = payload.get('timestamp', datetime.now().isoformat())
        
        # 提取關鍵數據
        reactor_data = {
            'temperature': float(payload.get('temp', 0)),
            'pressure': float(payload.get('pres', 0)),
            'coolant_flow': float(payload.get('flow', 0)),
            'coolant_inlet_temp': float(payload.get('t_in', 0)),
            'coolant_outlet_temp': float(payload.get('t_out', 0))
        }
        
        # 執行異常檢測
        process_sensor_data(reactor_data, timestamp, client)
        
    except json.JSONDecodeError:
        print(f"[ERROR] JSON 解析失敗：{msg.payload}")
    except Exception as e:
        print(f"[ERROR] 數據處理異常：{e}")

# ===== 異常檢測邏輯 =====
def process_sensor_data(reactor_data, timestamp, mqtt_client):
    """
    1. 執行孤立森林異常檢測
    2. 檢查固定閾值告警
    3. 發送相應級別的告警
    """
    
    # 孤立森林異常檢測
    features = np.array([
        reactor_data['temperature'],
        reactor_data['pressure'],
        reactor_data['coolant_flow'],
        reactor_data['temperature'] - (reactor_data['coolant_inlet_temp'] + reactor_data['coolant_outlet_temp']) / 2,
        reactor_data['coolant_flow'] / (reactor_data['pressure'] + 0.1)
    ]).reshape(1, -1)
    
    scaled = scaler.transform(features)
    anomaly_score = iso_forest.decision_function(scaled)[0]
    is_anomaly = iso_forest.predict(scaled)[0] == -1
    
    # 確定告警級別
    alert_level = None
    alert_reason = []
    
    if reactor_data['temperature'] > ALERT_LEVELS['CRITICAL']['temp_max']:
        alert_level = 'CRITICAL'
        alert_reason.append(f"溫度超臨界：{reactor_data['temperature']:.1f}°C")
    elif reactor_data['pressure'] > ALERT_LEVELS['CRITICAL']['pressure_max']:
        alert_level = 'CRITICAL'
        alert_reason.append(f"壓力超臨界：{reactor_data['pressure']:.1f} bar")
    elif reactor_data['temperature'] > ALERT_LEVELS['HIGH']['temp_max']:
        alert_level = 'HIGH'
        alert_reason.append(f"溫度過高：{reactor_data['temperature']:.1f}°C")
    elif is_anomaly and anomaly_score < -0.7:
        alert_level = 'MEDIUM'
        alert_reason.append(f"檢測到異常模式（分數：{anomaly_score:.3f}）")
    
    # 發送告警
    if alert_level:
        send_alert(alert_level, alert_reason, reactor_data, timestamp, mqtt_client)

# ===== 告警發送函式 =====
def send_alert(level, reasons, sensor_data, timestamp, mqtt_client):
    """
    根據告警級別採取不同的行動
    """
    alert_msg = {
        'timestamp': timestamp,
        'level': level,
        'reasons': reasons,
        'sensor_data': sensor_data
    }
    
    print(f"\n[ALERT-{level}] {timestamp}")
    print(f"  原因：{', '.join(reasons)}")
    
    if level == 'CRITICAL':
        print("  行動：自動啟動應急冷卻、關閉進料")
        # 發送指令至 PLC 觸發應急程序
        trigger_emergency_mitigation(mqtt_client)
        # 同時發送 SMS 與 Email
        send_sms_alert(ALERT_RECIPIENTS['sms'], alert_msg)
        send_email_alert(ALERT_RECIPIENTS['email'], alert_msg)
        
    elif level == 'HIGH':
        print("  行動：SMS 通知操作員")
        send_sms_alert(ALERT_RECIPIENTS['sms'], alert_msg)
        
    elif level == 'MEDIUM':
        print("  行動：Email 通知維護團隊")
        send_email_alert(ALERT_RECIPIENTS['email'], alert_msg)
    
    # 發布至 MQTT，供雲端系統記錄
    mqtt_client.publish(MQTT_PUBLISH_TOPIC, json.dumps(alert_msg))

def trigger_emergency_mitigation(mqtt_client):
    """
    發送指令至 PLC，觸發應急冷卻、關閉進料等
    """
    command = {
        'action': 'emergency_shutdown',
        'coolant_max_flow': 100,  # 冷卻系統全開
        'feed_stop': True
    }
    mqtt_client.publish('factory/reactor/commands', json.dumps(command))
    print("  已向 PLC 發送應急指令")

def send_sms_alert(phone_numbers, alert_msg):
    """
    透過 SMS 通知（需集成第三方 SMS 網關，例如 Twilio）
    """
    alert_text = f"[{alert_msg['level']}] {alert_msg['reasons'][0]} T={alert_msg['sensor_data']['temperature']:.0f}°C P={alert_msg['sensor_data']['pressure']:.0f}bar"
    for phone in phone_numbers:
        print(f"  SMS 已發送至 {phone}：{alert_text[:50]}...")

def send_email_alert(emails, alert_msg):
    """
    發送詳細的 Email 告警
    """
    subject = f"[{alert_msg['level']}] 反應釜異常告警 - {alert_msg['timestamp']}"
    body = f"""
    告警級別：{alert_msg['level']}
    時間：{alert_msg['timestamp']}
    
    異常原因：
    {chr(10).join(['  - ' + r for r in alert_msg['reasons']])}
    
    當前感測數據：
      溫度：{alert_msg['sensor_data']['temperature']:.1f}°C
      壓力：{alert_msg['sensor_data']['pressure']:.1f} bar
      冷卻流量：{alert_msg['sensor_data']['coolant_flow']:.1f} L/min
    
    請立即查看儀表板或聯繫值班人員。
    """
    
    # 使用 SMTP 發送（此處需配置郵件伺服器）
    for email in emails:
        print(f"  Email 已發送至 {email}")

# ===== 主程式 =====
if __name__ == '__main__':
    print("=" * 60)
    print("邊緣計算節點 - MQTT 異常檢測與告警服務")
    print("=" * 60)
    
    # 初始化 MQTT
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = on_message
    
    print(f"\n連接 MQTT Broker：{MQTT_BROKER}:{MQTT_PORT}")
    client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
    
    # 啟動客戶端迴圈
    print("監聽中...按 Ctrl+C 停止\n")
    client.loop_forever()

備註：上述代碼示例已簡化以供理解。實際部署時需補充：(1) 錯誤處理與日誌記錄、(2) 數據庫連接、(3) 真實的 SMS/Email 網關集成、(4) 與 PLC 控制邏輯的深度耦合。ATLANTIS 的技術團隊可提供完整的、已測試的生產級代碼。

最終結論與聯絡方式

經過九個完整章節的深度剖析，你已經掌握了「石化反應器 Physical AI 監控系統」的核心原理、成本構成、ROI 計算邏輯，以及選型決策框架。

如果你的企業正面臨以下挑戰：

• 停機成本高達每年 報價制，但缺乏預防手段
• 產品品質波動大，良率難以穩定在 98% 以上
• 需要滿足食品、製藥、化工等行業的法規稽核需求
• 擁有多座反應釜，希望透過智慧化實現跨釜優化

ATLANTIS 昶特正是你的合作夥伴。我們提供：

• ✅ 31 年本土製造經驗，理解台灣石化廠的實際痛點
• ✅ 完整的感測器、PLC、軟體、雲端一體化方案
• ✅ 分階段投資模式，降低初期成本與風險
• ✅ 專業的現場支持、員工培訓、持續優化
• ✅ 約 1.5 年的投資回本週期，之後年度淨收益 報價制

文章資訊
撰寫時間：2026 年 8 月｜更新時間：2026 年 8 月
作者：ATLANTIS 昶特應用工程團隊｜字數：22,500+ 字
適用對象：化工廠工程師、設施管理者、採購決策者、工廠主管
本文為 Physical AI 系列完整指南第一篇，後續將涵蓋「邊緣計算硬體選型」「雲端架構設計」「機器學習模型調教」等深度主題。