工業溫度計與壓力錶 AWS IoT 智慧監控平台完整指南｜實時異常檢測、成本優化、數據驅動管理

前言：為什麼需要 IoT 智慧監控平台？

在傳統製造業、能源管理、環保監控領域，工業溫度計與壓力錶是生產控制與安全防護的核心。然而，大多數企業仍採用：

• 人工定時抄錶 — 耗時、易錯、無法即時預警
• 單機監控 — 數據孤島、無法整合分析
• 事後維護 — 設備故障才發現，成本高昂

昶特 ATLANTIS 與 AWS IoT 的結合，將您的傳統儀表轉變為智慧監控生態：

據台灣製造業聯誼會 2024 年調查，導入 IoT 監控的企業：

• 設備故障率 下降 60-75%
• 能源成本 節省 12-18%
• 生產效率 提升 8-15%
• 平均投資回本週期 12-18 個月

第 2 章：實際應用場景與尖峰負荷分析

2.1 典型應用場景

場景 1：製造業 - 射出成型工廠

監測點：32 台成型機的模具溫度、液壓壓力、冷卻水溫

痛點：模具溫度波動導致成品不良率 8-12%、人工檢查 4 小時/班

解決方案：即時監控模具溫度 ±2°C，異常時自動調整冷卻系統，將不良率降至 1-2%

效益：年度節省（成品報廢 + 人工）約 NT$450,000

場景 2：能源管理 - 熱水鍋爐系統

監測點：鍋爐進出水溫、蒸汽壓力、回水溫度

痛點：燃料浪費 15-20%、高峰期壓力波動造成故障 2-3 次/月

解決方案：預測性維護、燃料消耗最佳化、壓力智慧調節

效益：燃料成本下降 12-15%，故障率下降 85%

場景 3：環保監控 - 污水處理廠

監測點：16 個池體溫度、曝氣壓力、進出口流量

痛點：微生物活性對溫度敏感，人工監測困難

解決方案：溫度區間自動控制、壓力動態調整、處理效能最佳化

效益：處理效率提升 18-22%，能耗下降 20%

2.2 尖峰負荷分析表

數據量分析：

• 日均數據產生：約 69,120 條（48 個點 × 24 小時 × 平均 60 秒間隔）
• 月均存儲需求：約 2.1 GB（壓縮後）
• AWS 費用估算：約 NT$800-1,200/月（含存儲、計算、傳輸）

第 4 章：昶特 ATLANTIS IoT 數據平台完整方案

4.1 為什麼選擇昶特 ATLANTIS？

4.2 核心產品與特性

產品 1：ATLANTIS IoT 數據收集模組

將傳統模擬信號（4-20mA、0-10V）轉換為數位信號，支援：

• 16 路通道獨立採集
• 每秒 100+ 筆數據處理能力
• 本地 SD 卡 32GB 存儲（斷網自動緩存）
• Ethernet、WiFi、LTE 多模連接
• IP65 防護，工業級可靠性

圖 1：ATLANTIS IoT 數據收集網關 — 支持多通道模擬信號數位化與實時邊緣處理

產品 2：ATLANTIS 智慧監控軟體（雲端版）

基於 AWS 雲平台的全功能監控應用，特性包括：

• 實時儀表板 — 即時顯示所有監測點狀態、趨勢圖表
• 多層告警 — 靜態閾值、動態異常、趨勢預測三種檢測方式
• 數據分析 — 跨時間段、跨設備的對比分析與統計報表
• 移動應用 — iOS/Android App，隨時隨地監控
• API 開放 — REST API 支援第三方應用整合

圖 2：ATLANTIS 監控儀表板 — 實時數據、異常告警、趨勢分析一覽無遺

4.3 昶特方案 vs 競品對比

4.4 部署方案

方案 A：快速導入版（1-2 週）

• 8 支溫度計 + 8 支壓力錶
• 1 台 IoT 網關
• 雲端監控軟體（年度授權）
• 基礎安裝與調試
• 投資回本週期：16-20 個月

方案 B：完整監控版（3-4 週）

• 24 支溫度計 + 24 支壓力錶
• 3 台 IoT 網關（容災備份）
• 進階監控軟體（含 AI 異常檢測）
• 移動應用部署
• 完整安裝、調試、員工培訓
• 投資回本週期：12-15 個月

方案 C：企業級解決方案（5-6 週）

• 48+ 監測點（儀表、感測器任意組合）
• 多網關冗余架構（N+1 容錯設計）
• 企業級軟體授權（多用戶、多角色）
• 進階分析（ML 模型、預測維護）
• 專業安裝、調試、培訓、定期維護
• 投資回本週期：8-12 個月

第 6 章：AWS IoT 整合程式碼範例

6.1 Python 設備端範例（MQTT 連接 AWS IoT Core）

此示例展示如何從昶特 IoT 網關讀取溫度、壓力數據，並通過 MQTT 發布至 AWS IoT Core。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
昶特 ATLANTIS IoT 網關 — AWS IoT Core MQTT 發布器
作用：每分鐘從本地網關收集溫度、壓力數據，通過 MQTT 發送至 AWS IoT Core
"""

import json
import time
import logging
from datetime import datetime
from awsiot import aws_iot_mqtt_client_builder

# 配置日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# AWS IoT 認證參數（需替換為實際值）
ENDPOINT = "xxxxxxxxxx.iot.ap-northeast-1.amazonaws.com"
CLIENT_ID = "atlantis-gateway-001"
PATH_TO_CERT = "/path/to/device-cert.pem.crt"
PATH_TO_KEY = "/path/to/device-key.pem.key"
PATH_TO_ROOT_CA = "/path/to/AmazonRootCA1.pem"
THING_NAME = "atlantis-factory-monitor"

# MQTT 主題設定
TEMPERATURE_TOPIC = f"factory/sensors/temperature"
PRESSURE_TOPIC = f"factory/sensors/pressure"
STATUS_TOPIC = f"factory/gateway/status"

def build_mqtt_client():
    """建立 AWS IoT MQTT 客戶端"""
    mqtt_client = aws_iot_mqtt_client_builder.MqttClientBuilder()\
        .with_endpoint(ENDPOINT)\
        .with_client_id(CLIENT_ID)\
        .with_cert_credentials_provider(
            certificate_path=PATH_TO_CERT,
            private_key_path=PATH_TO_KEY,
            ca_cert_path=PATH_TO_ROOT_CA
        )\
        .with_conn_ack_callback(on_connection_interrupted)\
        .with_on_message_received_callback(on_message_received)\
        .build()
    return mqtt_client

def read_sensor_data():
    """
    從昶特 IoT 網關讀取感測數據
    實務上，此函數應連接到網關的本地 API 或 Modbus/TCP
    """
    # 示例數據（實務上應從硬體讀取）
    sensors_data = {
        "gateway_id": "atlantis-gw-001",
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "temperature": [
            {"sensor_id": "TM-001", "value": 45.2, "unit": "°C"},
            {"sensor_id": "TM-002", "value": 38.5, "unit": "°C"},
            {"sensor_id": "TM-003", "value": 52.1, "unit": "°C"},
        ],
        "pressure": [
            {"sensor_id": "PM-001", "value": 6.8, "unit": "bar"},
            {"sensor_id": "PM-002", "value": 5.2, "unit": "bar"},
            {"sensor_id": "PM-003", "value": 7.5, "unit": "bar"},
        ]
    }
    return sensors_data

def publish_to_aws(mqtt_client, data):
    """發布數據至 AWS IoT Core"""
    try:
        # 發布溫度數據
        for temp_sensor in data["temperature"]:
            temp_payload = {
                "sensor_id": temp_sensor["sensor_id"],
                "timestamp": data["timestamp"],
                "value": temp_sensor["value"],
                "unit": temp_sensor["unit"]
            }
            mqtt_client.publish(
                topic=TEMPERATURE_TOPIC,
                qos=1,
                payload=json.dumps(temp_payload)
            )
            logger.info(f"發布溫度: {temp_sensor['sensor_id']} = {temp_sensor['value']}°C")

        # 發布壓力數據
        for pressure_sensor in data["pressure"]:
            pressure_payload = {
                "sensor_id": pressure_sensor["sensor_id"],
                "timestamp": data["timestamp"],
                "value": pressure_sensor["value"],
                "unit": pressure_sensor["unit"]
            }
            mqtt_client.publish(
                topic=PRESSURE_TOPIC,
                qos=1,
                payload=json.dumps(pressure_payload)
            )
            logger.info(f"發布壓力: {pressure_sensor['sensor_id']} = {pressure_sensor['value']} bar")

        # 發布網關狀態
        status_payload = {
            "gateway_id": data["gateway_id"],
            "timestamp": data["timestamp"],
            "status": "online",
            "temperature_count": len(data["temperature"]),
            "pressure_count": len(data["pressure"])
        }
        mqtt_client.publish(
            topic=STATUS_TOPIC,
            qos=1,
            payload=json.dumps(status_payload)
        )
        logger.info("網關狀態已發布")

    except Exception as e:
        logger.error(f"發布失敗: {str(e)}")

def on_connection_interrupted(connection, error, **kwargs):
    """MQTT 連線中斷處理"""
    logger.warning(f"MQTT 連線中斷: {error}")

def on_message_received(topic, payload, **kwargs):
    """接收 AWS 下行命令"""
    logger.info(f"收到命令 [{topic}]: {payload.decode('utf-8')}")

def main():
    """主程序"""
    logger.info("昶特 ATLANTIS - AWS IoT 連接器啟動")

    # 建立 MQTT 客戶端
    mqtt_client = build_mqtt_client()
    mqtt_client.connect()

    logger.info("已連接至 AWS IoT Core")

    # 訂閱下行命令主題（可選）
    # mqtt_client.subscribe(topic="factory/commands/#", qos=1, callback=on_message_received)

    # 主循環：每分鐘讀取並發布一次數據
    try:
        while True:
            data = read_sensor_data()
            publish_to_aws(mqtt_client, data)
            logger.info(f"下次更新時間: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time() + 60))}")
            time.sleep(60)  # 每 60 秒更新一次
    except KeyboardInterrupt:
        logger.info("程式終止")
    finally:
        mqtt_client.disconnect()

if __name__ == "__main__":
    main()

6.2 AWS Lambda 函數範例（異常檢測）

此 Lambda 函數接收 MQTT 消息，進行簡單的異常檢測，若超出閾值則發送告警。

import json
import boto3
import logging
from datetime import datetime

# 初始化 AWS 服務
dynamodb = boto3.resource('dynamodb')
sns = boto3.client('sns')
cloudwatch = boto3.client('cloudwatch')

# DynamoDB 表格
sensor_data_table = dynamodb.Table('AtlantisTemperaturePressureData')
alert_table = dynamodb.Table('AtlantisAlerts')

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 異常檢測閾值
THRESHOLDS = {
    "TM-001": {"min": 35, "max": 55},  # 溫度計 TM-001: 35-55°C
    "TM-002": {"min": 30, "max": 50},  # 溫度計 TM-002: 30-50°C
    "TM-003": {"min": 40, "max": 60},  # 溫度計 TM-003: 40-60°C
    "PM-001": {"min": 5, "max": 8},    # 壓力錶 PM-001: 5-8 bar
    "PM-002": {"min": 4, "max": 7},    # 壓力錶 PM-002: 4-7 bar
    "PM-003": {"min": 6, "max": 9},    # 壓力錶 PM-003: 6-9 bar
}

# SNS 主題 ARN（用於發送通知）
SNS_TOPIC_ARN = "arn:aws:sns:ap-northeast-1:123456789:atlantis-alerts"

def detect_anomaly(sensor_id, value):
    """
    檢測異常
    回傳: (is_anomaly, anomaly_type)
    - anomaly_type: 'normal', 'high', 'low', 'unknown_sensor'
    """
    if sensor_id not in THRESHOLDS:
        return True, "unknown_sensor"

    threshold = THRESHOLDS[sensor_id]

    if value > threshold["max"]:
        return True, "high"
    elif value < threshold["min"]:
        return True, "low"
    else:
        return False, "normal"

def send_alert(sensor_id, value, anomaly_type, unit):
    """發送告警"""
    timestamp = datetime.utcnow().isoformat() + "Z"

    # 1. 保存至 DynamoDB
    alert_record = {
        "alert_id": f"{sensor_id}#{timestamp}",
        "sensor_id": sensor_id,
        "value": value,
        "anomaly_type": anomaly_type,
        "unit": unit,
        "timestamp": timestamp,
        "status": "unread"
    }
    alert_table.put_item(Item=alert_record)

    # 2. 發送 SNS 通知
    message = f"""
異常告警 - 昶特 ATLANTIS 監控系統

感測器: {sensor_id}
值: {value} {unit}
異常類型: {anomaly_type}
時間: {timestamp}

請立即檢查設備狀況！
監控平台: https://dashboard.atlantis.iot/alerts
    """

    try:
        sns.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject=f"⚠️ 異常告警: {sensor_id}",
            Message=message
        )
        logger.info(f"告警已發送: {sensor_id}")
    except Exception as e:
        logger.error(f"告警發送失敗: {str(e)}")

    # 3. 發送 CloudWatch 指標
    cloudwatch.put_metric_data(
        Namespace='AtlantisMonitoring',
        MetricData=[
            {
                'MetricName': 'AnomalyCount',
                'Value': 1,
                'Unit': 'Count',
                'Dimensions': [
                    {'Name': 'SensorID', 'Value': sensor_id},
                    {'Name': 'AnomalyType', 'Value': anomaly_type}
                ]
            }
        ]
    )

def save_sensor_data(sensor_id, value, unit, timestamp):
    """保存感測數據至 DynamoDB"""
    try:
        sensor_data_table.put_item(
            Item={
                "sensor_id#timestamp": f"{sensor_id}#{timestamp}",
                "sensor_id": sensor_id,
                "value": value,
                "unit": unit,
                "timestamp": timestamp,
                "ttl": int(datetime.utcnow().timestamp()) + (90 * 24 * 3600)  # 90 天後過期
            }
        )
        logger.info(f"數據已保存: {sensor_id} = {value}")
    except Exception as e:
        logger.error(f"數據保存失敗: {str(e)}")

def lambda_handler(event, context):
    """
    Lambda 入口函數

    事件格式:
    {
        "sensor_id": "TM-001",
        "timestamp": "2024-01-15T10:30:45Z",
        "value": 58.5,
        "unit": "°C"
    }
    """
    try:
        sensor_id = event.get('sensor_id')
        value = event.get('value')
        unit = event.get('unit')
        timestamp = event.get('timestamp', datetime.utcnow().isoformat() + "Z")

        logger.info(f"處理數據: {sensor_id} = {value} {unit}")

        # 1. 保存數據
        save_sensor_data(sensor_id, value, unit, timestamp)

        # 2. 異常檢測
        is_anomaly, anomaly_type = detect_anomaly(sensor_id, value)

        # 3. 如果異常，發送告警
        if is_anomaly:
            logger.warning(f"異常檢測: {sensor_id} = {value} ({anomaly_type})")
            send_alert(sensor_id, value, anomaly_type, unit)

        # 4. 返回響應
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Data processed',
                'sensor_id': sensor_id,
                'is_anomaly': is_anomaly,
                'anomaly_type': anomaly_type
            })
        }

    except Exception as e:
        logger.error(f"Lambda 執行失敗: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

6.3 JSON 設備配置檔案

以下示例展示昶特 IoT 網關的配置檔案格式，定義所有連接的感測器、告警閾值、資料發送頻率等。

{
  "gateway": {
    "id": "atlantis-gw-001",
    "name": "工廠 A 監控網關",
    "location": "台北市北投區致遠一路二段 109 號",
    "version": "1.2.3",
    "connection": {
      "type": "ethernet",
      "ip": "192.168.1.100",
      "gateway": "192.168.1.1",
      "dns": ["8.8.8.8", "8.8.4.4"]
    },
    "mqtt": {
      "broker": "xxxxxxxxxx.iot.ap-northeast-1.amazonaws.com",
      "port": 8883,
      "protocol": "mqtts",
      "client_id": "atlantis-gateway-001",
      "keep_alive": 60,
      "publish_interval": 60
    }
  },
  "sensors": {
    "temperature": [
      {
        "channel": 1,
        "sensor_id": "TM-001",
        "name": "射出機 #1 模具溫度",
        "type": "ATLANTIS TM-02",
        "input_type": "4-20mA",
        "min_range": 0,
        "max_range": 150,
        "unit": "°C",
        "decimals": 1,
        "enabled": true,
        "alert": {
          "enabled": true,
          "threshold_min": 35,
          "threshold_max": 55,
          "threshold_delta": 2.0,
          "alert_level_low": "warning",
          "alert_level_high": "critical"
        }
      },
      {
        "channel": 2,
        "sensor_id": "TM-002",
        "name": "射出機 #2 模具溫度",
        "type": "ATLANTIS TM-02",
        "input_type": "4-20mA",
        "min_range": 0,
        "max_range": 150,
        "unit": "°C",
        "decimals": 1,
        "enabled": true,
        "alert": {
          "enabled": true,
          "threshold_min": 30,
          "threshold_max": 50,
          "threshold_delta": 2.0,
          "alert_level_low": "warning",
          "alert_level_high": "critical"
        }
      },
      {
        "channel": 3,
        "sensor_id": "TM-003",
        "name": "冷卻水進水溫度",
        "type": "ATLANTIS TM-01",
        "input_type": "4-20mA",
        "min_range": -20,
        "max_range": 100,
        "unit": "°C",
        "decimals": 1,
        "enabled": true,
        "alert": {
          "enabled": true,
          "threshold_min": 15,
          "threshold_max": 25,
          "threshold_delta": 1.0,
          "alert_level_low": "info",
          "alert_level_high": "warning"
        }
      }
    ],
    "pressure": [
      {
        "channel": 9,
        "sensor_id": "PM-001",
        "name": "液壓系統壓力",
        "type": "ATLANTIS PM-02",
        "input_type": "4-20mA",
        "min_range": 0,
        "max_range": 25,
        "unit": "bar",
        "decimals": 2,
        "enabled": true,
        "alert": {
          "enabled": true,
          "threshold_min": 5,
          "threshold_max": 8,
          "threshold_delta": 0.3,
          "alert_level_low": "warning",
          "alert_level_high": "critical"
        }
      },
      {
        "channel": 10,
        "sensor_id": "PM-002",
        "name": "冷卻水循環壓力",
        "type": "ATLANTIS PM-01",
        "input_type": "4-20mA",
        "min_range": 0,
        "max_range": 10,
        "unit": "bar",
        "decimals": 2,
        "enabled": true,
        "alert": {
          "enabled": true,
          "threshold_min": 3,
          "threshold_max": 6,
          "threshold_delta": 0.2,
          "alert_level_low": "warning",
          "alert_level_high": "warning"
        }
      }
    ]
  },
  "notifications": {
    "email": {
      "enabled": true,
      "recipients": ["ops@factory-a.com", "manager@factory-a.com"],
      "alert_levels": ["critical", "warning"]
    },
    "sms": {
      "enabled": true,
      "recipients": ["+886912345678"],
      "alert_levels": ["critical"]
    },
    "webhook": {
      "enabled": true,
      "url": "https://api.factory-a.com/webhooks/alerts",
      "method": "POST",
      "timeout": 10
    }
  },
  "data_export": {
    "format": "csv",
    "retention_days": 90,
    "archive_to_s3": true,
    "s3_bucket": "atlantis-data-archive-ap-northeast-1"
  }
}