數位壓力錶接入數位孿生完整指南：以 AT-803 建虛擬世界、做預測、打造儀表板與戰情室

前言：數位孿生的第一塊地基，是一支「說真話」的數位壓力錶

談到數位孿生，多數人腦中浮現的是三維模型、會動的管線動畫、牆面大螢幕上的即時 KPI，以及「AI 預測什麼時候會壞」。這些畫面都對，但它們全部建立在同一個前提上：虛擬世界裡的數字，必須真實反映現場。只要底層感測資料有偏差，模型越精密，錯得越有自信；戰情室畫面越漂亮，誤導的範圍越大。

數位孿生的定義其實不複雜：它是實體資產在數位世界的對應體，透過持續的資料同步，讓我們能監看現況、模擬「如果這樣調整會怎樣」、預測未來狀態，並把決策回饋到現場。它和一般 3D 展示模型最大的差別，在於「持續同步」與「可以拿來做決策」。而同步的來源，就是散布在現場的感測器——其中，壓力與壓差是流體系統裡資訊量最高的物理量之一。

為什麼壓力是孿生最重要的訊號之一

水、空氣、蒸汽、冷媒、液壓油、製程氣體、潔淨室的空氣，本質上都是流體。流體系統的健康狀態，幾乎都會先寫在壓力上：濾網開始堵塞，前後壓差會一天一天上升；泵浦葉輪磨耗，同樣流量下的揚程會下滑；管網出現微漏，停機保壓時壓力會緩慢下降；潔淨室或手術室的門關不密，房間壓差會被拉低。換句話說，一支擺在正確位置、讀值可信的數位壓力錶，就是孿生模型判斷「系統現在好不好」最直接的證據。

這也是為什麼在智慧建築、製造產線、風力發電、航太維修、半導體與醫療院所，孿生專案的第一階段幾乎都會回到同一份清單：哪些點位要量壓力？量程多大？要多準？訊號怎麼送出來？多久校一次？

孿生專案最常被忽略的風險：感測器漂移

任何感測器都會隨時間、溫度循環與壓力循環慢慢漂移。漂移本身並不可怕，可怕的是孿生系統分不出「是現場真的變了，還是感測器變了」。舉例來說，一支微差壓傳送器的零點每個月慢慢偏移，孿生模型看到的是「房間壓差持續上升」或「濾網提早老化」，於是預測模型提前發出更換工單；更糟的情況是漂移方向剛好抵銷真實變化，讓真正的異常被掩蓋。久而久之，現場人員發現「戰情室常常亂叫」，整套系統就失去信任。

解法不是買更多感測器，而是在系統裡放進量測真值錨點：用精度等級明顯更高、並可追溯校正的標準級儀錶，定期比對線上感測器，把誤差、修正係數與不確定度寫回孿生的感測器模型。如此一來，孿生不只知道「讀值是多少」，還知道「這個讀值可以信到什麼程度」。

AT-803 在本文中的定位

昶特 ATLANTIS 的 AT-803 高精度數位壓力錶（產品目錄型號 AT803-D，高精度數位差壓錶）官方公開精度可達 ±0.02%F.S.，定位為現場與實驗室壓力量測、校準等高精度微壓量測的標準級儀錶。在數位孿生架構裡，它最大的價值不是「多一個點位」，而是成為整個壓力資料鏈的參考基準：校準潔淨室、手術室、空調風管、排氣系統等微壓點位的線上傳送器，量化每個點位的不確定度，並協助孿生分辨感測器漂移與真實製程變化。

第 2 章　AT-803 數位壓力錶在孿生系統的角色：量測真值錨點

一套數位孿生裡會有上百、上千個壓力點位，但它們的「身分」並不相同。有些點位負責 24 小時連續回報，有些點位由巡檢人員定期抽查，還有少數儀錶的任務是當作參考標準，去檢查其他點位有沒有說錯話。把角色分清楚，才能決定每個點位要用什麼儀錶、什麼訊號、多久校正一次。

2.1　AT-803 產品定位

真值錨點｜微壓標準級

AT-803 高精度數位壓力錶（目錄型號 AT803-D）

圖說：ATLANTIS AT-803（AT803-D）高精度數位差壓錶 — 精度可達 ±0.02%F.S.，適用現場與實驗室的高精度微壓量測與校準。

官方公開規格與定位

• 精度：可達 ±0.02%F.S.（昶特官網公開值）。
• 定位：高精度數位差壓錶，微壓量測的標準級儀錶。
• 用途：現場、實驗室壓力量測與校準等高精度微壓量測。
• 操作：依產品照片，面板具大尺寸液晶顯示，以及設定、單位切換、歸零與背光按鍵。
• 量程、接頭與輸出介面：依訂購規格而定，請於詢價時提供使用情境，由昶特確認最適配置。

差壓錶的低壓端通大氣時，即可量測「相對大氣」的微壓，因此 AT-803 同時適合量測房間室壓、風管靜壓等錶壓型微壓，以及濾網、門縫兩側的差壓。

完整校正級產品請見校正專用精密壓力錶系列。

2.2　數位壓力錶在孿生中的三種角色

我們建議在點位規劃時，就替每一支儀錶標上角色。角色決定了訊號形式、資料頻率與校正策略：

表 2-1　三種角色與對應的昶特儀錶

三種角色之間是互補關係：線上節點提供「頻率」，巡檢節點提供「覆蓋」，真值錨點提供「可信度」。很多專案只投資第一種，結果資料量很大、可信度不明，最後預測模型被迫加上很寬的容許帶，失去提早預警的意義。

2.3　為什麼孿生需要真值錨點：漂移與真實變化長得很像

下圖是一個示意情境：某潔淨室的真實室壓差長期維持在 15 Pa，但負責回報的線上傳送器每月緩慢漂移約 0.3 Pa（假設值）。沒有參考標準的系統，會在第 5 個月看到讀值越過 16.5 Pa 的警報門檻，於是發出誤報，甚至讓預測模型以為送風或回風系統出了問題；以 AT-803 每 4 個月比對一次並寫回修正係數的系統，讀值偏差始終控制在 1.2 Pa 以內。

這個例子凸顯三件事。第一，漂移是「慢」的，日常看不出來，但會被趨勢模型放大。第二，預測模型無法自己分辨漂移與真實變化，必須有外部參考。第三，錨點比對的價值不只是「修正讀值」，而是把每一次比對的誤差歷史留下來，讓孿生知道每個點位的漂移速度，進而決定合理的校正週期。

2.4　TUR 4:1 與不確定度：讓「可信程度」變成數字

校正時常用的經驗法則是測試不確定度比（TUR）至少 4:1：參考標準的誤差，應小於被校儀錶允許誤差的四分之一。以同量程計算，AT-803 的 ±0.02%F.S. 對上 ±0.1%F.S. 等級的儀錶，比值為 5:1；對上 ±0.5%F.S. 等級的線上微差壓傳送器，比值可達 25:1。若量程不同，必須先換算成相同單位的絕對誤差再比較，這也是為什麼詢價時要提供被校儀錶的量程與精度等級。

表 2-2　微壓比對的簡化不確定度預算（示意）

把各來源轉成標準不確定度後，以平方和開根號合成，再乘以涵蓋因子 k（常用 k = 2，約 95% 信賴水準）即為擴充不確定度：

孿生系統只要多存一個欄位「U（k=2）」，戰情室就能在畫面上顯示「15.2 ± 0.2 Pa」而不只是「15.2 Pa」，警報門檻也能據此設定合理的緩衝帶，大幅降低誤報。第 8 章的範例三提供了計算修正係數與 TUR 的程式。

2.5　AT-803 接入孿生的三條路徑

AT-803 的輸出介面依訂購規格而定。實務上可以依需求選擇以下三條路徑之一，或混合使用：

不論採用哪條路徑，請在詢價時說明使用情境（量程、介質、安裝方式、是否需要訊號輸出、是否需要校正報告），由昶特業務確認可供應的規格。

2.6　微壓量測的八個現場要點

AT-803 的價值建立在正確的量測方法上。微壓（數 Pa 到數百 Pa）的量測非常敏感，以下八點是比對作業最常出錯的地方：

1. 先歸零再量測：兩端連通、讀值穩定後再歸零，並在比對結束時複查零點。
2. 導壓管同高：高低壓端的導壓管保持同一高度，避免管內空氣溫差造成的浮力誤差。
3. 管路氣密：檢查接頭、軟管是否老化或彎折，微小漏氣就會讓讀值偏低。
4. 熱平衡：儀錶從冷氣房帶到現場，先放置一段時間再量測。
5. 關門與避風：量測房間壓差時關閉門窗，取壓口避開出風口與人員動線。
6. 時間平均：微壓讀值會隨氣流跳動，以數十秒到數分鐘的平均值作為比對值。
7. 同點比對：參考標準與被校儀錶接在同一個取壓點或用三通並聯，避免兩點本身就有差異。
8. 完整記錄：記錄日期、時間、溫度、門窗狀態、比對點、讀值與人員，讓紀錄能被稽核。

2.7　昶特 ATLANTIS 數位孿生量測層產品配置

以下產品圖片與描述皆取自昶特官網產品目錄。每張卡片標示該儀錶在孿生中的建議角色，實際量程、接頭、輸出與防護等級依訂購規格而定。

真值錨點｜中高壓

DPG-0.05B4.5(5CD) 高精度數位壓力錶

採用進口精密壓力感測器與內部溫度補償感測器，精度可達 ±0.05%F.S.，可作為標準校驗用數位壓力錶。孿生用途：液壓、空壓、製程管線等中高壓點位的比對基準。

真值錨點｜迴路校驗

IPC-801 智能壓力校驗儀

可量測壓力、電壓與電流，適用校準壓力錶、精密壓力錶、壓力與差壓傳送器、壓力調節閥及壓力開關。孿生用途：從感測器、4–20 mA 迴路到系統讀值的整條量測鏈驗證。

線上節點｜微壓

DMPT-S221 數位微差壓傳送器

採用進口微差壓感測元件，精度高、長期穩定性好，可選 RS485 或 4–20 mA 輸出；適用通風、消防排煙、風機檢測與空調過濾系統。孿生用途：室壓、風管靜壓、濾網壓差的連續監測。

巡檢／現場顯示｜微壓

DMPG-X022 數位微差壓錶

進口微差壓感測元件搭配超低功耗電路，安裝方式與機械式微差壓錶相同；適用潔淨室、手術室、無塵室、通風系統與風機測試。孿生用途：門口現場顯示，讓人員與戰情室看到同一個數字。

線上節點｜差壓

SDPT-3351 智慧型差壓傳送器

5 位數液晶顯示、可選配 HART，最高精度可達 ±0.1%F.S.，具警報、溫度指示、溫度補償與資料備份恢復功能。孿生用途：過濾器、換熱器、泵浦進出口差壓。

線上節點｜防腐差壓

DPT-AC HART 防腐型差壓傳送器

SS316L 膜片式感測器，高精度與高抗蝕；HART 通訊支援遠端組態與診斷，適用石油、化工、電子、空調、能源與廢水處理。孿生用途：腐蝕性或關鍵介質的差壓點位。

線上節點｜管線壓力

SDPT-3100 智能型壓力傳送器

以微處理器為核心，具靈活的壓力校準與輸出、HART 協議通訊、環境溫度自動補償與多種參數設定。孿生用途：給水、冰水、空壓、液壓與蒸汽管線壓力。

巡檢節點

DPG-X112 高精度藍芽數位壓力錶（可旋轉式）

錶頭可旋轉 330°，主副屏顯示壓力、環境溫度、最大／最小值與量程，支援 Type-C 或藍芽匯出記錄。孿生用途：尚未佈線點位的巡檢資料，帶時間戳上傳。

線上節點｜溫度補償

STT HART 智能型溫度傳送器

一體化溫度傳送器，可搭配熱電阻或熱電偶，HART 支援遠端組態與診斷。孿生用途：油品黏度、氣體密度與熱平衡計算都需要溫度，壓力孿生通常要配對溫度點。

第 4 章　儀表板與戰情室：通用 BI、工業監控型、製造現場型怎麼搭配

孿生平台負責「理解與推演」，儀表板與戰情室負責「讓人看懂並採取行動」。很多專案的失敗，不是孿生做不出來，而是畫面做得像報表、警報多到沒人理、異常發生時不知道該找誰。依用途，儀表板軟體可分為三類，各有擅長的時間尺度與使用者。

4.1　通用 BI 工具：跨部門彙總與管理報表

• Power BI：兼具報表能力與相對親民的導入門檻，是製造業常用的選擇，並與 Microsoft Fabric 的資料與即時分析服務整合。適合月報、能源報表、跨廠比較。
• Tableau：擅長進階視覺分析與探索式分析，適合分析師深入挖掘壓力、能耗與品質之間的關聯。
• Metabase：開源選項，適合有自建能力的團隊快速建立內部查詢與報表。

BI 工具的強項是彙總與比較，弱項是秒級時序資料與即時警報。壓力資料進 BI 前，建議先在資料層彙總成 5 分鐘、1 小時或日平均，並帶上資料可用率，避免報表把「沒有資料」誤當成「數值為零」。

4.2　工業監控型：時序資料與即時警報

• Grafana：在工業監控與時序資料視覺化上特別突出，開源、資料來源支援廣，技術團隊常用來處理 IoT 感測資料；搭配時序資料庫可做到秒級趨勢與警報規則。第 8 章範例六提供 TimescaleDB 與 Grafana 的查詢範例。
• Siemens Insights Hub（原 MindSphere）：強調把工業資產連接到儀表板，支援廠區層級的 KPI 與警示，適合已採用 Siemens 自動化生態系的工廠。

4.3　製造現場型：從警報直接走到工單

• Fabrico：把 OEE 與維修（CMMS）資料整合在同一個平台，異常可直接產生工單並推播給技術人員，縮短「發現問題」到「有人去修」的時間。
• FactoryStudio（Tatsoft，FrameworX 平台）：提供警報管理、儀表板設計器與資料視覺化，定位於工業即時應用，並內建 OPC UA、MQTT 與 Sparkplug B 等通訊能力，適合做 SCADA 等級的現場畫面。

表 4-1　三類儀表板工具比較

實務上很少只用一種。常見組合是：Grafana 或 FactoryStudio 做現場即時畫面，Fabrico 或既有 CMMS 處理工單，Power BI 做管理報表，三者共用同一份點位主檔與時序資料。

4.4　戰情室設計七原則

1. 三層畫面：總覽（全廠健康）→ 區域（系統或樓層）→ 設備（單點趨勢與歷史），任何異常三次點擊內可以追到源頭。
2. 以例外為導向：依 ISA-101 高效能人機介面的精神，正常狀態用低彩度呈現，只有異常才上色，讓眼睛自然被問題吸走。
3. 警報合理化：依 ISA-18.2／IEC 62682 為每個警報定義原因、後果、處置與優先度；ISA-18.2 以每位操作員每小時約 6 個警報作為「很可能可接受」的參考量級。
4. 每個警報都有主人：警報要連到處置程序與負責人，最好能一鍵產生工單，處置結果再回寫到孿生作為模型訓練資料。
5. 資料品質可視化：校正逾期、通訊中斷、超量程要有獨立的顏色或圖示，與「製程異常」分開，否則值班人員會在錯的地方找問題。
6. 時間同步：所有閘道與伺服器以 NTP 對時，資料一律以 UTC 儲存、以當地時間顯示，事件追溯時才對得上。
7. 戰情室是流程，不只是螢幕：值班表、交接紀錄、事件檢討與 KPI 月會，才讓畫面上的數字變成持續改善。

4.5　戰情室總覽畫面示意

以下為一個以壓力資料為主的設施戰情室總覽示意（數值皆為示意）。注意畫面上同時呈現「製程狀態」與「資料可信度」：

圖 4-1　以壓力資料為核心的戰情室總覽示意（示意畫面，非實際系統數據）

4.6　壓力類 KPI 定義表

表 4-2　戰情室常用的壓力類 KPI

第 6 章　預測怎麼做：從趨勢外推、物理模型到機器學習

「預測」常被誤解成一定要用深度學習。事實上，壓力類的預測大多可以從簡單的方法開始，逐步升級。重點不是方法多先進，而是預測結果能不能被現場相信、能不能轉成工單。以下依複雜度由低到高介紹五種方法，以及一個常被忽略的關鍵：漂移感知。

6.1　規則與門檻：最簡單，也最容易被誤用

固定門檻（例如「壓差大於 250 Pa 就警報」）是所有系統的起點。它的問題在於兩端：門檻設太緊，壓力稍有波動就警報；設太鬆，等到警報響時已經來不及。改善方法有三：加上遲滯帶（例如超過 250 Pa 觸發、降到 240 Pa 以下才解除），加上延時確認（持續超過 60 秒才成立），以及把門檻與不確定度綁在一起（讀值加上 U（k=2）仍在範圍內才算合格）。

6.2　統計管制：抓出「慢慢變壞」

對於緩慢變化，指數加權移動平均（EWMA）與累積和管制（CUSUM）比固定門檻敏感得多。EWMA 會讓最近的資料權重較高，同時平滑掉瞬間雜訊；CUSUM 則累積每一次偏離目標值的量，小偏差持續發生時會很快觸發。這兩種方法計算量小，可以直接在邊緣閘道或時序資料庫中執行。

6.3　趨勢外推：濾網、過濾器的剩餘天數

濾網、濾芯與過濾器的壓差上升通常相當規律，非常適合用線性或分段線性回歸外推「哪一天會到更換門檻」。實務上有兩個關鍵：

• 先正規化再外推：壓差會隨流量變化。若風機或泵浦有變頻，必須把壓差換算到同一個基準流量，否則趨勢線會被運轉模式牽著走。纖維濾材多接近層流，壓差約與流量成正比；紊流的管路元件則較接近與流量平方成正比，指數最好用現場資料回歸校估。
• 用日平均而不是原始值：以每日同一運轉狀態下的平均壓差做回歸，可以大幅降低雜訊，讓剩餘天數穩定，不會每小時跳來跳去。

第 8 章範例四提供了完整的剩餘天數計算程式，可以直接用在空調箱濾網、FFU、空壓過濾器與潤滑油濾芯。

6.4　物理＋資料的混合模型：先算「應該是多少」

混合模型的做法，是先用物理關係算出「在這個運轉條件下，壓力應該是多少」，再拿實測值比較，對殘差做異常偵測。常見的物理關係包括：

• 泵浦：依泵浦性能曲線，由流量與轉速推算應有揚程；實測揚程（出口減入口壓力換算）持續低於預期，代表葉輪磨耗或內部洩漏。
• 管路與閥件：壓降與流量的關係（紊流近似與流量平方成正比），阻力係數上升代表結垢、堵塞或閥門開度異常。
• 房間壓力平衡：送風量減去回風與排氣量，再扣掉門縫與結構的滲漏，決定室壓；送排風量穩定但室壓下降，就要懷疑門或氣密。

這正是模擬型平台（例如 Ansys Twin Builder 的 Hybrid Analytics）擅長的事，但即使沒有大型平台，也可以用試算表或簡單程式先建立物理基準。混合模型對量測準確度特別敏感，因為殘差本身就是「實測減預期」：實測端的系統誤差會直接變成殘差偏移，這也是校正錨點在混合模型中不可或缺的原因。

6.5　機器學習：適合多變數、難以寫成公式的情境

當一個系統有數十個彼此影響的變數（例如潔淨室的多個房間壓差、送排風量、門禁事件與天候），孤立森林、自編碼器等非監督式方法可以學習「正常的組合長什麼樣子」，偏離時提出警示。導入時請留意：

• 訓練資料必須先排除已知的感測器故障與校正逾期資料，否則模型會把壞資料學成「正常」。
• 要能解釋：戰情室需要知道「是哪幾個點讓模型覺得異常」，否則值班人員無從下手。
• 要能回饋：每一次警報的處置結果（真異常、誤報、儀錶問題）都要標記回去，讓模型持續修正。

6.6　漂移感知：讓預測模型知道「感測器也會變」

這是本文最想強調的一點。預測模型通常假設感測器是準的，但現場的感測器會漂移。漂移感知的做法，是把感測器本身也當成孿生的一部分：

1. 每個點位帶著不確定度：從點位主檔讀取最近一次比對的修正係數與 U（k=2），所有預測與警報都以修正後讀值與不確定度計算。
2. 辨識漂移特徵：單一感測器緩慢、單調地偏離，而相鄰點位與物理模型都沒有對應變化，優先懷疑感測器本身。
3. 自動排程比對：孿生判定「疑似漂移」時，不直接派維修工單，而是先產生「AT-803 比對」工單，確認後再決定是修設備還是修儀錶。
4. 以歷史調整週期：累積每個點位的歷次比對誤差，漂移慢的點位延長週期、漂移快的點位縮短週期，校正資源用在最需要的地方。

6.7　預測模型上線前的驗收指標

表 6-1　預測與警報模型的驗收指標

第 8 章　客戶自行串接範例程式：從數位壓力錶到孿生與戰情室

六段範例依資料流向排列：範例一、二在邊緣層把感測訊號轉成帶品質碼的資料；範例三把 AT-803 的比對結果變成修正係數與不確定度；範例四做濾網剩餘天數預測；範例五把狀態寫進 Azure Digital Twins；範例六在時序資料庫與 Grafana 建立戰情室畫面。各段可以獨立使用，也可以串成一條完整的管線。

表 8-1　六段範例程式一覽

8.1　範例一：RS-485 Modbus RTU 輪詢 → MQTT（含品質碼與校正中繼資料）

這支程式是整條管線的起點。它和一般「讀值就送」的範例有三個差異：讀值先套用 AT-803 比對所得的修正係數；每筆資料帶上品質碼（通訊錯誤、超量程、校正逾期）；時間戳一律使用 UTC。適用於 DMPT-S221 等具 RS485 輸出的傳送器，暫存器位址請以各型號通訊手冊為準。

# ===== 檔案名稱：modbus_to_mqtt_edge.py =====
# ===== 用途：RS-485 Modbus RTU 輪詢壓力／差壓傳送器 → 發佈 MQTT（含資料品質與校正中繼資料）=====
# ===== 執行環境：Python 3.9+，工業電腦或邊緣閘道（Linux／Windows 皆可）=====
# ===== 安裝套件：pip install "pymodbus[serial]>=3.6" "paho-mqtt>=2.0"（serial 會一併安裝 pyserial）=====
# ===== 注意：站號、暫存器位址、資料型別、比例係數皆為示意，請以各型號通訊手冊為準 =====
# ===== 注意：本程式僅供監測用途；安全連鎖與控制請保留在 PLC／DCS／SIS =====

import json
import struct
import time
import logging
from datetime import datetime, timezone

from pymodbus.client import ModbusSerialClient
import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("edge")

# ---------- 1. 通訊設定 ----------
SERIAL_PORT = "/dev/ttyUSB0"     # Windows 範例："COM3"
BAUDRATE = 9600
PARITY = "N"
STOPBITS = 1
POLL_SECONDS = 5                 # 輪詢週期（秒）

MQTT_HOST = "192.168.10.20"      # 廠內 MQTT Broker 位址（請改為貴公司設定）
MQTT_PORT = 1883                 # 正式環境建議改用 8883 並啟用 TLS 與帳號密碼
MQTT_TOPIC_ROOT = "plant1/fmcs"  # 主題根目錄（示意）

# ---------- 2. 點位主檔：孿生需要的中繼資料 ----------
# slave：Modbus 站號；register：保持暫存器起始位址；dtype：int16 或 float32；scale：讀值乘數
# cal_gain／cal_offset／u95／cal_due：由 AT-803 比對作業產生（見範例三）
POINTS = [
    {
        "point_id": "CR-B2-ROOM-DP-01",
        "slave": 1, "register": 0, "dtype": "int16", "scale": 0.1,
        "unit": "Pa", "range_min": 0.0, "range_max": 50.0,
        "location": "B2 黃光區 → 走廊",
        "cal_gain": 0.99673, "cal_offset": -0.1295,
        "u95": 0.076, "cal_due": "2027-04-12",
    },
    {
        "point_id": "AHU-03-FILTER-DP",
        "slave": 2, "register": 0, "dtype": "float32", "scale": 1.0,
        "unit": "Pa", "range_min": 0.0, "range_max": 500.0,
        "location": "3F 空調箱 AHU-03 中效濾網",
        "cal_gain": 1.0, "cal_offset": 0.0,
        "u95": 2.5, "cal_due": "2027-01-31",
    },
]


def call_with_unit_id(func, unit_id, **kwargs):
    """pymodbus 3.x 不同版本的站號參數名稱不同（device_id／slave／unit），依序嘗試。"""
    for key in ("device_id", "slave", "unit"):
        try:
            return func(**kwargs, **{key: unit_id})
        except TypeError:
            continue
    raise RuntimeError("無法呼叫 pymodbus 讀取函式，請確認套件版本")


def decode(registers, dtype):
    """把暫存器內容轉成數值。float32 預設高字在前，若讀值異常請交換兩個暫存器順序。"""
    if dtype == "int16":
        v = registers[0]
        return v - 65536 if v > 32767 else v
    if dtype == "float32":
        raw = struct.pack(">HH", registers[0], registers[1])
        return struct.unpack(">f", raw)[0]
    raise ValueError(f"未支援的資料型別：{dtype}")


def read_point(client, p):
    """讀取單一點位，回傳（修正後讀值, 品質碼）。"""
    count = 2 if p["dtype"] == "float32" else 1
    try:
        rr = call_with_unit_id(client.read_holding_registers, p["slave"],
                               address=p["register"], count=count)
    except Exception as exc:                      # 通訊逾時、斷線等
        log.error("%s 讀取例外：%s", p["point_id"], exc)
        return None, "COMM_ERROR"
    if rr.isError():
        return None, "COMM_ERROR"

    raw = decode(rr.registers, p["dtype"]) * p["scale"]
    value = raw * p["cal_gain"] + p["cal_offset"]   # 套用 AT-803 比對所得的修正係數

    span = p["range_max"] - p["range_min"]
    today = datetime.now(timezone.utc).date().isoformat()
    if value < p["range_min"] - 0.02 * span or value > p["range_max"] + 0.02 * span:
        quality = "OUT_OF_RANGE"
    elif today > p["cal_due"]:
        quality = "CAL_OVERDUE"                   # 讀值仍送出，但標示校正逾期
    else:
        quality = "GOOD"
    return round(value, 4), quality


def build_mqtt_client():
    try:                                          # paho-mqtt 2.x
        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="edge-fmcs-01")
    except AttributeError:                        # paho-mqtt 1.x
        c = mqtt.Client(client_id="edge-fmcs-01")
    c.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
    c.loop_start()
    return c


def main():
    modbus = ModbusSerialClient(port=SERIAL_PORT, baudrate=BAUDRATE, parity=PARITY,
                                stopbits=STOPBITS, bytesize=8, timeout=1)
    if not modbus.connect():
        raise SystemExit("RS-485 連線失敗：請檢查序列埠名稱、接線極性與終端電阻")
    mq = build_mqtt_client()
    log.info("邊緣輪詢開始，共 %d 個點位", len(POINTS))
    try:
        while True:
            t0 = time.time()
            ts = datetime.now(timezone.utc).isoformat()   # 一律以 UTC 儲存，顯示時再轉當地時間
            for p in POINTS:
                value, quality = read_point(modbus, p)
                payload = {
                    "ts": ts, "point_id": p["point_id"], "value": value,
                    "unit": p["unit"], "quality": quality, "u95": p["u95"],
                    "cal_due": p["cal_due"], "location": p["location"],
                }
                topic = f"{MQTT_TOPIC_ROOT}/{p['point_id']}"
                mq.publish(topic, json.dumps(payload, ensure_ascii=False), qos=1)
                if quality != "GOOD":
                    log.warning("%s 品質=%s 讀值=%s", p["point_id"], quality, value)
            time.sleep(max(0.0, POLL_SECONDS - (time.time() - t0)))
    except KeyboardInterrupt:
        log.info("使用者中止")
    finally:
        modbus.close()
        mq.loop_stop()
        mq.disconnect()


if __name__ == "__main__":
    main()

8.2　範例二：4–20 mA 換算與 NAMUR NE43 故障判斷

類比訊號最大的風險，是「斷線時讀到 0 mA，換算成負壓或零壓」，讓孿生誤以為現場失壓。NAMUR NE43 定義了故障電流範圍，依此判斷可以把「迴路或感測器故障」與「製程異常」分開處理。

# ===== 檔案名稱：ma_scaling_ne43.py =====
# ===== 用途：4–20 mA 換算為壓力工程值，並依 NAMUR NE43 區分「製程異常」與「迴路／感測器故障」=====
# ===== 執行環境：Python 3.9+（只用標準函式庫）=====
# ===== mA 值來源：PLC 類比輸入、資料擷取模組或 Modbus 類比模組（依貴公司架構）=====

from dataclasses import dataclass


@dataclass
class AnalogPoint:
    point_id: str
    range_min: float   # 4 mA 對應的工程值
    range_max: float   # 20 mA 對應的工程值
    unit: str


# NAMUR NE43 常用門檻（mA）
NE43_FAIL_LOW = 3.6     # ≤ 3.6 mA：判定故障（斷線、感測器失效）
NE43_MEAS_LOW = 3.8     # 3.6～3.8 mA：低於量測範圍
NE43_MEAS_HIGH = 20.5   # 20.5～21.0 mA：高於量測範圍
NE43_FAIL_HIGH = 21.0   # ≥ 21.0 mA：判定故障


def ma_to_value(ma: float, p: AnalogPoint):
    """回傳（工程值, 品質碼）。故障時不回傳數值，避免孿生把故障當成真實壓力。"""
    if ma <= NE43_FAIL_LOW:
        return None, "SENSOR_FAIL_LOW"
    if ma >= NE43_FAIL_HIGH:
        return None, "SENSOR_FAIL_HIGH"
    value = p.range_min + (ma - 4.0) * (p.range_max - p.range_min) / 16.0
    if ma < NE43_MEAS_LOW:
        return round(value, 4), "UNDER_RANGE"
    if ma > NE43_MEAS_HIGH:
        return round(value, 4), "OVER_RANGE"
    return round(value, 4), "GOOD"


if __name__ == "__main__":
    pump = AnalogPoint("CHW-P02-DISCHARGE", 0.0, 10.0, "bar")   # 0–10 bar 對應 4–20 mA（示意）
    for ma in (2.0, 3.7, 4.0, 12.0, 19.9, 20.8, 22.0):
        value, quality = ma_to_value(ma, pump)
        print(f"{ma:5.2f} mA → {value} {pump.unit}　品質={quality}")

執行結果示意：12.00 mA 對應 5.0 bar、品質 GOOD；20.80 mA 對應 10.5 bar、品質 OVER_RANGE；2.00 mA 與 22.00 mA 不回傳數值，分別標示為 SENSOR_FAIL_LOW 與 SENSOR_FAIL_HIGH。

8.3　範例三：以 AT-803 為參考標準，計算修正係數、TUR 與不確定度

這是 AT-803 在孿生中最核心的用法。比對作業完成後，把每個測試點的參考讀值與被校讀值填入，程式會算出修正前的最大誤差與遲滯、最小平方法修正係數、TUR 與擴充不確定度，並產生可直接寫入點位主檔的 JSON 校正紀錄。

# ===== 檔案名稱：at803_reference_check.py =====
# ===== 用途：以 AT-803（官方公開精度 ±0.02%F.S.）為參考標準，比對線上微差壓傳送器，
#             計算修正前誤差、遲滯、修正係數、TUR 與擴充不確定度，輸出可寫入孿生的校正紀錄 =====
# ===== 執行環境：Python 3.9+（只用標準函式庫）=====
# ===== 注意：數據為示意；正式校正請依貴公司程序書，參考標準本身須定期送校並保留報告 =====

import json
import math
from datetime import date, timedelta

# ---------- 1. 被校儀錶與參考標準 ----------
DUT = {                      # Device Under Test：被校的線上感測器
    "point_id": "CR-B2-ROOM-DP-01",
    "range_fs": 50.0,        # 滿量程（Pa）
    "unit": "Pa",
    "spec_pct_fs": 0.5,      # 允許誤差 ±0.5%F.S.（示意，請依實際規格）
}
REF = {                      # 參考標準：AT-803（請以其校正報告的量程與不確定度為準）
    "model": "ATLANTIS AT-803 (AT803-D)",
    "range_fs": 50.0,        # 假設與被校儀錶同量程（示意）
    "spec_pct_fs": 0.02,
    "cert_no": "REF-CERT-XXXX",
}
CAL_INTERVAL_DAYS = 180      # 下次比對間隔，可依歷次漂移調整

# ---------- 2. 比對數據：(方向, 參考讀值, 被校讀值) ----------
# up = 升壓、down = 降壓；每個點建議取 3 次以上讀值平均後再填入
READINGS = [
    ("up", 0.00, 0.09), ("up", 12.50, 12.66), ("up", 25.00, 25.21),
    ("up", 37.50, 37.73), ("up", 50.00, 50.26),
    ("down", 50.00, 50.29), ("down", 37.50, 37.78), ("down", 25.00, 25.27),
    ("down", 12.50, 12.71), ("down", 0.00, 0.12),
]


def linear_fit(xs, ys):
    """最小平方法 y = a·x + b（x = 被校讀值，y = 參考讀值）；修正後讀值 = a·讀值 + b"""
    n = len(xs)
    mx, my = sum(xs) / n, sum(ys) / n
    sxx = sum((x - mx) ** 2 for x in xs)
    sxy = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
    a = sxy / sxx
    return a, my - a * mx


def main():
    fs = DUT["range_fs"]
    errors = [(d, ref, dut, dut - ref) for d, ref, dut in READINGS]
    max_err = max(abs(e[3]) for e in errors)

    # 遲滯：同一參考點升壓與降壓讀值的最大差
    hyst = 0.0
    for ref in sorted({r for _, r, _ in READINGS}):
        ups = [dut for d, r, dut in READINGS if d == "up" and r == ref]
        downs = [dut for d, r, dut in READINGS if d == "down" and r == ref]
        if ups and downs:
            hyst = max(hyst, abs(ups[0] - downs[0]))

    gain, offset = linear_fit([e[2] for e in errors], [e[1] for e in errors])
    residual_max = max(abs(gain * e[2] + offset - e[1]) for e in errors)

    # TUR：被校儀錶允許誤差 ÷ 參考標準誤差（換算成相同單位）
    dut_tol = DUT["spec_pct_fs"] / 100 * fs
    ref_err = REF["spec_pct_fs"] / 100 * REF["range_fs"]
    tur = dut_tol / ref_err

    # 簡化不確定度：參考標準＋修正後殘差＋遲滯（皆以矩形分佈換算），k = 2
    u_ref = ref_err / math.sqrt(3)
    u_resid = residual_max / math.sqrt(3)
    u_hyst = (hyst / 2) / math.sqrt(3)
    u95 = 2 * math.sqrt(u_ref ** 2 + u_resid ** 2 + u_hyst ** 2)

    record = {
        "point_id": DUT["point_id"],
        "date": date.today().isoformat(),
        "reference": {"model": REF["model"], "cert_no": REF["cert_no"]},
        "as_found": {
            "max_error": round(max_err, 3),
            "max_error_pct_fs": round(max_err / fs * 100, 3),
            "hysteresis": round(hyst, 3),
            "pass": max_err <= dut_tol,
        },
        "correction": {"gain": round(gain, 5), "offset": round(offset, 4),
                       "residual_max": round(residual_max, 4)},
        "tur": round(tur, 1),
        "u95": round(u95, 3),
        "unit": DUT["unit"],
        "cal_due": (date.today() + timedelta(days=CAL_INTERVAL_DAYS)).isoformat(),
    }
    print(json.dumps(record, ensure_ascii=False, indent=2))

    if tur < 4:
        print("提醒：TUR 低於 4:1，請改用量程更貼近或等級更高的參考標準")
    if not record["as_found"]["pass"]:
        print("提醒：修正前誤差超出允許值，請評估上次比對後的歷史資料是否需標記")


if __name__ == "__main__":
    main()

以範例數據執行，修正前最大誤差為 0.29 Pa（0.58%F.S.），超出 ±0.5%F.S. 的允許值，程式會提醒評估歷史資料；修正係數約為增益 0.99673、偏移 −0.1295 Pa，修正後殘差最大約 0.058 Pa，TUR 為 25:1，擴充不確定度約 0.076 Pa（k = 2）。範例一的點位主檔即採用這組結果。這個例子也說明了：若沒有定期比對，這支傳送器的偏差會一直被當成「真實室壓」進入孿生。

8.4　範例四：濾網剩餘天數預測（含風量正規化）

對應第 6 章的趨勢外推方法。輸入每日平均壓差與風量，程式先把壓差換算到設計風量，再以最近 21 天回歸計算上升速率與剩餘天數，可直接接到工單系統或戰情室的「濾網剩餘天數」磚。

# ===== 檔案名稱：filter_remaining_days.py =====
# ===== 用途：以壓差趨勢外推濾網剩餘天數（含風量正規化），供戰情室排程與自動開立工單 =====
# ===== 執行環境：Python 3.9+（只用標準函式庫）=====
# ===== 輸入：每日同一運轉狀態下的平均壓差與平均風量（可由時序資料庫彙總後取得）=====

from datetime import date, timedelta
from statistics import mean, pstdev

DP_LIMIT = 250.0        # 更換門檻（Pa），依濾網規格與 SOP 設定（示意）
Q_DESIGN = 12000.0      # 設計風量（m³/h）
FLOW_EXPONENT = 1.0     # ΔP ∝ Qⁿ：纖維濾材接近層流 n≈1；粗效或紊流元件可能接近 2，請以實測回歸校估
WINDOW_DAYS = 21        # 回歸使用的最近天數


def normalize(dp, q):
    """把不同風量下的壓差換算到設計風量，避免變頻造成的假趨勢。"""
    return dp * (Q_DESIGN / q) ** FLOW_EXPONENT


def fit_line(xs, ys):
    """簡單線性回歸，回傳（斜率, 截距, 殘差標準差）。"""
    mx, my = mean(xs), mean(ys)
    sxx = sum((x - mx) ** 2 for x in xs)
    slope = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sxx
    intercept = my - slope * mx
    residuals = [y - (slope * x + intercept) for x, y in zip(xs, ys)]
    return slope, intercept, pstdev(residuals)


def remaining_days(history):
    """history：[(日期, 日平均壓差 Pa, 日平均風量 m³/h), ...]，建議至少 7 天。"""
    recent = history[-WINDOW_DAYS:]
    xs = [(d - recent[0][0]).days for d, _, _ in recent]
    ys = [normalize(dp, q) for _, dp, q in recent]
    slope, intercept, sd = fit_line(xs, ys)
    current = slope * xs[-1] + intercept
    if slope <= 0:
        return None, current, slope, sd
    return max(0.0, (DP_LIMIT - current) / slope), current, slope, sd


if __name__ == "__main__":
    # 示意資料：設計風量下的壓差每日約上升 1.6 Pa；實際風量因變頻在 9,000～12,000 m³/h 之間變動
    start = date(2026, 9, 15)
    history = []
    for i in range(30):
        q = 12000 - (i % 4) * 1000
        dp_at_design = 160 + 1.6 * i + (-1) ** i * 1.5        # 設計風量下的「真實」壓差
        dp_measured = dp_at_design * (q / Q_DESIGN) ** FLOW_EXPONENT
        history.append((start + timedelta(days=i), dp_measured, q))

    days_left, current, slope, sd = remaining_days(history)
    print(f"目前正規化壓差：{current:.1f} Pa｜上升速率：{slope:.2f} Pa/天｜殘差標準差：{sd:.1f} Pa")
    if days_left is None:
        print("壓差沒有上升趨勢，暫無更換需求（也請確認感測器是否卡值或通訊中斷）")
    else:
        due = history[-1][0] + timedelta(days=round(days_left))
        print(f"預估剩餘天數：{days_left:.0f} 天 → 建議於 {due.isoformat()} 前完成更換排程")

以範例資料執行，正規化後的目前壓差約 206 Pa、每日上升約 1.6 Pa，距離 250 Pa 門檻約剩 27 天。若未做風量正規化，變頻造成的壓差起伏會讓回歸斜率大幅失真。

8.5　範例五：DTDL 模型與 Azure Digital Twins 狀態更新

以下 DTDL 定義了「受控房間」與「壓力感測點」兩種模型，感測點透過 monitors 關係連到房間。感測點除了壓力值，還帶有品質碼、不確定度、修正係數與校正到期日，讓孿生在查詢時就能知道每個讀值的可信程度。

[
  {
    "@id": "dtmi:atlantisdemo:Room;1",
    "@type": "Interface",
    "@context": "dtmi:dtdl:context;2",
    "displayName": "受控房間（示意）",
    "contents": [
      { "@type": "Property", "name": "roomName", "schema": "string" },
      { "@type": "Property", "name": "pressureSetpoint", "schema": "double" },
      { "@type": "Property", "name": "readiness", "schema": "string" }
    ]
  },
  {
    "@id": "dtmi:atlantisdemo:PressureSensor;1",
    "@type": "Interface",
    "@context": "dtmi:dtdl:context;2",
    "displayName": "壓力／差壓感測點（示意）",
    "contents": [
      { "@type": ["Property", "Pressure"], "name": "pressure", "schema": "double", "unit": "pascal" },
      { "@type": "Property", "name": "quality", "schema": "string" },
      { "@type": "Property", "name": "u95", "schema": "double" },
      { "@type": "Property", "name": "calDue", "schema": "date" },
      { "@type": "Property", "name": "calGain", "schema": "double" },
      { "@type": "Property", "name": "calOffset", "schema": "double" },
      { "@type": "Relationship", "name": "monitors", "target": "dtmi:atlantisdemo:Room;1" }
    ]
  }
]

# ===== 檔案名稱：adt_update_twin.py =====
# ===== 用途：把邊緣閘道送出的壓力讀值與校正中繼資料，更新到 Azure Digital Twins 的對應孿生 =====
# ===== 安裝套件：pip install azure-digitaltwins-core azure-identity =====
# ===== 前置作業（由客戶自行完成）：建立 Azure Digital Twins 執行個體、上傳 DTDL 模型、
#       設定本程式身分的存取權限（例如 Azure Digital Twins Data Owner 角色）=====
# ===== 建議：高頻原始資料先存時序資料庫，孿生只更新「最新狀態」或分鐘級彙總，避免超出服務限制 =====

import json
import os

from azure.core.exceptions import ResourceNotFoundError
from azure.digitaltwins.core import DigitalTwinsClient
from azure.identity import DefaultAzureCredential

ADT_URL = os.environ.get("ADT_URL", "https://<your-instance>.api.<region>.digitaltwins.azure.net")
SENSOR_MODEL = "dtmi:atlantisdemo:PressureSensor;1"

client = DigitalTwinsClient(ADT_URL, DefaultAzureCredential())


def ensure_twin(twin_id: str, cal: dict):
    """孿生不存在時建立；屬性名稱必須與上傳的 DTDL 模型一致。"""
    try:
        client.get_digital_twin(twin_id)
    except ResourceNotFoundError:
        client.upsert_digital_twin(twin_id, {
            "$metadata": {"$model": SENSOR_MODEL},
            "pressure": 0.0,
            "quality": "INIT",
            "u95": cal.get("u95", 0.0),
            "calDue": cal.get("cal_due", "2000-01-01"),
            "calGain": cal.get("cal_gain", 1.0),
            "calOffset": cal.get("cal_offset", 0.0),
        })


def update_from_payload(payload: dict):
    """payload 格式與範例一的 MQTT 訊息相同。"""
    twin_id = payload["point_id"]
    ensure_twin(twin_id, payload)
    patch = [
        {"op": "replace", "path": "/quality", "value": payload["quality"]},
        {"op": "replace", "path": "/u95", "value": payload["u95"]},
        {"op": "replace", "path": "/calDue", "value": payload["cal_due"]},
    ]
    if payload.get("value") is not None:        # 感測器故障時不覆寫最後一筆有效讀值
        patch.append({"op": "replace", "path": "/pressure", "value": payload["value"]})
    client.update_digital_twin(twin_id, patch)


def link_sensor_to_room(sensor_id: str, room_id: str):
    """建立「感測器 monitors 房間」的關係，讓孿生能由房間反查負責的感測點。"""
    relationship_id = f"{sensor_id}-monitors-{room_id}"
    client.upsert_relationship(sensor_id, relationship_id, {
        "$relationshipId": relationship_id,
        "$sourceId": sensor_id,
        "$relationshipName": "monitors",
        "$targetId": room_id,
    })


if __name__ == "__main__":
    demo = {"point_id": "CR-B2-ROOM-DP-01", "value": 14.82, "quality": "GOOD",
            "u95": 0.076, "cal_due": "2027-04-12"}
    update_from_payload(demo)
    print(json.dumps(client.get_digital_twin(demo["point_id"]), ensure_ascii=False, indent=2))

實務建議：Azure Digital Twins 適合保存「最新狀態」與「關係」，秒級的原始資料建議先寫入時序資料庫（範例六），再以分鐘級頻率更新孿生；身分驗證請使用受控識別或服務主體，不要把金鑰寫在程式中。

8.6　範例六：TimescaleDB 時序資料與 Grafana 戰情室查詢

最後一段建立壓力時序資料表與 5 分鐘連續彙總，並提供三個 Grafana 面板查詢：區域室壓趨勢、各點位資料可用率，以及最新讀值與品質。彙總時只計入品質良好的資料，同時保留樣本數，讓戰情室能分辨「數值正常」與「其實沒有資料」。

-- ===== 檔案名稱：timescale_grafana.sql =====
-- ===== 用途：建立壓力時序資料表（含品質碼與不確定度）、5 分鐘連續彙總，以及 Grafana 面板查詢 =====
-- ===== 環境：PostgreSQL 14+ 與 TimescaleDB 擴充套件；MQTT 訊息可由 Telegraf、Node-RED 或自寫程式寫入 =====

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE IF NOT EXISTS pressure_ts (
    ts        TIMESTAMPTZ      NOT NULL,
    point_id  TEXT             NOT NULL,
    value     DOUBLE PRECISION,                -- 感測器故障時為 NULL
    unit      TEXT             NOT NULL,
    quality   TEXT             NOT NULL,       -- GOOD / OUT_OF_RANGE / CAL_OVERDUE / COMM_ERROR ...
    u95       DOUBLE PRECISION                 -- 擴充不確定度（k = 2）
);
SELECT create_hypertable('pressure_ts', 'ts', if_not_exists => TRUE);
CREATE INDEX IF NOT EXISTS idx_pressure_point_ts ON pressure_ts (point_id, ts DESC);

-- 5 分鐘連續彙總：平均、最大、最小只計入品質良好的資料，並保留樣本數計算資料可用率
CREATE MATERIALIZED VIEW IF NOT EXISTS pressure_5m
WITH (timescaledb.continuous) AS
SELECT time_bucket(INTERVAL '5 minutes', ts) AS bucket,
       point_id,
       avg(CASE WHEN quality = 'GOOD' THEN value END)      AS avg_value,
       max(CASE WHEN quality = 'GOOD' THEN value END)      AS max_value,
       min(CASE WHEN quality = 'GOOD' THEN value END)      AS min_value,
       sum(CASE WHEN quality = 'GOOD' THEN 1 ELSE 0 END)   AS good_samples,
       count(*)                                            AS all_samples
FROM pressure_ts
GROUP BY bucket, point_id
WITH NO DATA;

SELECT add_continuous_aggregate_policy('pressure_5m',
       start_offset      => INTERVAL '1 day',
       end_offset        => INTERVAL '5 minutes',
       schedule_interval => INTERVAL '5 minutes');

-- Grafana 時序面板：B2 區室壓趨勢（$__timeFilter 為 Grafana PostgreSQL 資料來源巨集）
SELECT bucket AS "time", point_id AS metric, avg_value
FROM pressure_5m
WHERE $__timeFilter(bucket) AND point_id LIKE 'CR-B2-%'
ORDER BY bucket;

-- Grafana 表格面板：各點位資料可用率（%），由低到高排序，最需要處理的排最前面
SELECT point_id,
       round(100.0 * sum(good_samples) / nullif(sum(all_samples), 0), 2) AS availability_pct
FROM pressure_5m
WHERE $__timeFilter(bucket)
GROUP BY point_id
ORDER BY availability_pct ASC;

-- Grafana 表格面板：最新一筆讀值與品質（戰情室「資料品質」磚）
SELECT DISTINCT ON (point_id) point_id, ts AS "time", value, unit, quality, u95
FROM pressure_ts
WHERE ts > now() - INTERVAL '15 minutes'
ORDER BY point_id, ts DESC;

第 10 章　延伸閱讀：數位壓力錶選型、校正與數位化應用

本文聚焦在「數位壓力錶如何進入數位孿生」。若您正在比較型號、規劃校正制度，或想進一步了解特定產業的壓力量測，以下站內文章與工具可以接著閱讀：