Python：邊界網關數據採集與預處理

import json
import time
import serial
import struct
from datetime import datetime
from collections import deque
import numpy as np

class GasLineMonitor:
    """
    LNG 管線監控系統
    監控對象：壓力、流速
    目標：60 秒內檢測爆管
    """

    def __init__(self, pressure_port='/dev/ttyUSB0', flow_port='/dev/ttyUSB1'):
        self.pressure_serial = serial.Serial(pressure_port, 9600, timeout=2.0)
        self.flow_serial = serial.Serial(flow_port, 9600, timeout=2.0)

        # 歷史數據（60 個數據點 = 60 秒）
        self.pressure_history = deque(maxlen=60)
        self.flow_history = deque(maxlen=60)

        # 基線值（正常情況）
        self.baseline_pressure = None  # bar
        self.baseline_flow = None      # m³/h
        self.is_calibrated = False

        # 爆管判斷閾值
        self.pressure_drop_threshold = 0.3  # bar（壓力下降超過 0.3 bar）
        self.flow_drop_threshold = 25       # %（流速下降超過 25%）
        self.pressure_variance_threshold = 0.15  # 15% 壓力波動

    def read_pressure(self):
        """
        讀取壓力值（Modbus RTU）
        返回：bar（標準氣壓單位）
        """
        try:
            self.pressure_serial.write(
                bytes([0x01, 0x03, 0x00, 0x00, 0x00, 0x02, 0xC4, 0x0B])
            )
            response = self.pressure_serial.read(7)

            if len(response) < 7:
                return None

            # 解析浮點數
            raw_bytes = response[3:7]
            pressure_bar = struct.unpack('>f', raw_bytes)[0]

            self.pressure_history.append(pressure_bar)
            return pressure_bar

        except Exception as e:
            print(f"壓力讀取錯誤：{str(e)}")
            return None

    def read_flow_rate(self):
        """
        讀取流量值（渦街式流量計）
        返回：m³/h
        """
        try:
            self.flow_serial.write(
                bytes([0x02, 0x03, 0x00, 0x10, 0x00, 0x02, 0x74, 0x1A])
            )
            response = self.flow_serial.read(7)

            if len(response) < 7:
                return None

            flow_m3h = struct.unpack('>f', response[3:7])[0]

            self.flow_history.append(flow_m3h)
            return flow_m3h

        except Exception as e:
            print(f"流量讀取錯誤：{str(e)}")
            return None

    def calibrate_baseline(self):
        """
        校準基線值（首次啟動或定期校準）
        需要 60 秒穩定數據
        """
        if len(self.pressure_history) >= 60 and len(self.flow_history) >= 60:
            self.baseline_pressure = float(np.mean(list(self.pressure_history)))
            self.baseline_flow = float(np.mean(list(self.flow_history)))
            self.is_calibrated = True
            print(f"✓ 基線校準完成：壓力 {self.baseline_pressure:.2f} bar，流速 {self.baseline_flow:.2f} m³/h")
            return True
        return False

    def get_telemetry(self):
        """
        採集完整遙測數據（每秒一次）
        """
        pressure = self.read_pressure()
        flow = self.read_flow_rate()

        telemetry = {
            'timestamp': int(time.time()),
            'iso_timestamp': datetime.utcnow().isoformat() + 'Z',
            'pressure_bar': round(pressure, 2) if pressure else None,
            'flow_m3h': round(flow, 1) if flow else None,
            'sensor_status': 'OK' if (pressure and flow) else 'ERROR',
            'history_length': len(self.pressure_history)
        }

        # 如果還沒校準，嘗試校準
        if not self.is_calibrated:
            if self.calibrate_baseline():
                telemetry['calibration_status'] = 'COMPLETE'
            else:
                telemetry['calibration_status'] = 'IN_PROGRESS'

        return telemetry

def lambda_handler(event, context):
    """邊界採集 Lambda"""
    monitor = GasLineMonitor()

    # 連續讀取 60 秒
    for i in range(60):
        telemetry = monitor.get_telemetry()
        print(json.dumps(telemetry))

        # 發送到 AWS IoT Core MQTT
        # mqtt_client.publish('gas-line/telemetry', json.dumps(telemetry))

        time.sleep(1)

    return {
        'statusCode': 200,
        'message': '60秒監控完成'
    }

AWS IoT Events Detector Model

{
  "detectorModelName": "GasPipelineBurstDetector",
  "detectorModelDefinition": {
    "states": [
      {
        "stateName": "NORMAL",
        "onInput": {
          "transitionEvents": [
            {
              "eventName": "DetectPressureDrop",
              "condition": "isActive($variable.pressure_drops)",
              "actions": [
                {
                  "setVariable": {
                    "variableName": "pressure_drops",
                    "value": "$input.GaslineInput.pressure_bar < ($variable.baseline_pressure - 0.3)"
                  }
                },
                {
                  "setVariable": {
                    "variableName": "drop_timestamp",
                    "value": "timestamp()"
                  }
                }
              ],
              "nextState": "PRESSURE_DROP_DETECTED"
            },
            {
              "eventName": "DetectFlowDrop",
              "condition": "($input.GaslineInput.flow_m3h < ($variable.baseline_flow * 0.75))",
              "nextState": "FLOW_DROP_DETECTED"
            }
          ],
          "stateChangeEvents": []
        },
        "onEnter": {
          "events": [
            {
              "eventName": "EnterNormal",
              "actions": [
                {
                  "setVariable": {
                    "variableName": "pressure_drops",
                    "value": "false"
                  }
                }
              ]
            }
          ]
        }
      },
      {
        "stateName": "PRESSURE_DROP_DETECTED",
        "onInput": {
          "transitionEvents": [
            {
              "eventName": "ConfirmBurst",
              "condition": "($input.GaslineInput.flow_m3h < ($variable.baseline_flow * 0.60)) &&
                           (timestamp() - $variable.drop_timestamp < 30000)",
              "actions": [
                {
                  "sns": {
                    "targetArn": "arn:aws:sns:tw-south-1:123456789012:gas-burst-alert-critical",
                    "payload": {
                      "location": "$input.GaslineInput.sensor_id",
                      "pressure_bar": "$input.GaslineInput.pressure_bar",
                      "flow_m3h": "$input.GaslineInput.flow_m3h",
                      "alert_level": "CRITICAL_BURST",
                      "detection_time": "timestamp()"
                    }
                  }
                },
                {
                  "lambda": {
                    "functionArn": "arn:aws:lambda:tw-south-1:123456789012:function:BurstLocationCalculator"
                  }
                }
              ],
              "nextState": "BURST_CONFIRMED"
            },
            {
              "eventName": "FalseAlarm",
              "condition": "(timestamp() - $variable.drop_timestamp > 60000) &&
                           ($input.GaslineInput.pressure_bar > ($variable.baseline_pressure - 0.1))",
              "nextState": "NORMAL"
            }
          ]
        }
      },
      {
        "stateName": "FLOW_DROP_DETECTED",
        "onInput": {
          "transitionEvents": [
            {
              "eventName": "ConfirmWithPressure",
              "condition": "($input.GaslineInput.pressure_bar < ($variable.baseline_pressure - 0.2))",
              "actions": [
                {
                  "sns": {
                    "targetArn": "arn:aws:sns:tw-south-1:123456789012:gas-burst-alert-critical"
                  }
                },
                {
                  "dynamodb": {
                    "tableName": "GasBurstEvents",
                    "payload": {
                      "PK": "$input.GaslineInput.sensor_id",
                      "SK": "timestamp()",
                      "event_type": "BURST_CONFIRMED",
                      "pressure": "$input.GaslineInput.pressure_bar",
                      "flow": "$input.GaslineInput.flow_m3h"
                    }
                  }
                }
              ],
              "nextState": "BURST_CONFIRMED"
            }
          ]
        }
      },
      {
        "stateName": "BURST_CONFIRMED",
        "onInput": {
          "transitionEvents": [
            {
              "eventName": "MonitorRecovery",
              "condition": "timestamp() - $variable.drop_timestamp > 300000",
              "nextState": "NORMAL"
            }
          ]
        },
        "onExit": {
          "events": [
            {
              "eventName": "ExitBurst",
              "actions": [
                {
                  "sns": {
                    "targetArn": "arn:aws:sns:tw-south-1:123456789012:gas-burst-resolved",
                    "payload": {
                      "message": "爆管處理完成，管線恢復正常"
                    }
                  }
                }
              ]
            }
          ]
        }
      }
    ],
    "initialStateName": "NORMAL"
  }
}

Python：爆管位置推算與自動派遣

import json
import boto3
import math
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
sns = boto3.client('sns')
lambda_client = boto3.client('lambda')

# 管線配置（示例：台北市主幹道）
PIPELINE_SEGMENTS = {
    'MAIN_001': {'start_lat': 25.0330, 'start_lon': 121.5654, 'end_lat': 25.0410, 'end_lon': 121.5750, 'length_km': 1.2},
    'MAIN_002': {'start_lat': 25.0410, 'start_lon': 121.5750, 'end_lat': 25.0520, 'end_lon': 121.5890, 'length_km': 2.1},
    # ... 更多管線段
}

SENSOR_LOCATIONS = {
    'SENSOR_001': {'segment': 'MAIN_001', 'position': 0.3, 'lat': 25.0360, 'lon': 121.5690},
    'SENSOR_002': {'segment': 'MAIN_001', 'position': 0.9, 'lat': 25.0390, 'lon': 121.5730},
    'SENSOR_003': {'segment': 'MAIN_002', 'position': 0.5, 'lat': 25.0445, 'lon': 121.5800},
    # ... 更多感測器
}

REPAIR_TEAMS = {
    'TEAM_01': {'base_lat': 25.0400, 'base_lon': 121.5700, 'capacity': 5},
    'TEAM_02': {'base_lat': 25.0350, 'base_lon': 121.5650, 'capacity': 5},
    # ... 更多維修團隊
}

def calculate_burst_location(sensor_id, pressure_at_burst, flow_at_burst):
    """
    根據壓力降和流量推算爆管位置
    使用三角波前推進法（Pressure Wave Propagation）
    """
    if sensor_id not in SENSOR_LOCATIONS:
        return None

    sensor_info = SENSOR_LOCATIONS[sensor_id]
    segment_id = sensor_info['segment']
    segment_info = PIPELINE_SEGMENTS[segment_id]

    # 聲速（LNG 在管內約 400 m/s）
    sound_speed = 400  # m/s

    # 壓降波前距離估算
    # 壓力降 = ρ * c * ΔV
    # 其中 ρ 是密度，c 是聲速，ΔV 是速度變化
    pressure_wave_distance = (0.7 * sound_speed) * 60  # 60秒內傳播距離（米）

    # 推算爆管位置
    if flow_at_burst < (flow_at_burst * 0.6):  # 流速大幅下降
        # 爆管在上游（逆向傳播）
        estimated_distance_from_sensor = -pressure_wave_distance
    else:
        # 爆管在下游
        estimated_distance_from_sensor = pressure_wave_distance

    burst_position_km = sensor_info['position'] + (estimated_distance_from_sensor / 1000)

    # 計算地理坐標
    segment_length = segment_info['length_km']
    segment_ratio = burst_position_km / segment_length if segment_length > 0 else 0.5

    burst_lat = segment_info['start_lat'] + (segment_info['end_lat'] - segment_info['start_lat']) * segment_ratio
    burst_lon = segment_info['start_lon'] + (segment_info['end_lon'] - segment_info['start_lon']) * segment_ratio

    return {
        'segment_id': segment_id,
        'burst_position_km': round(burst_position_km, 2),
        'burst_latitude': round(burst_lat, 6),
        'burst_longitude': round(burst_lon, 6),
        'confidence': 0.85  # 準確度 85%
    }

def find_nearest_repair_team(burst_location):
    """
    找到最近的維修團隊並計算到達時間
    """
    burst_lat = burst_location['burst_latitude']
    burst_lon = burst_location['burst_longitude']

    min_distance = float('inf')
    nearest_team = None

    for team_id, team_info in REPAIR_TEAMS.items():
        distance = math.sqrt(
            (burst_lat - team_info['base_lat'])**2 +
            (burst_lon - team_info['base_lon'])**2
        ) * 111  # 轉換為公里（1 度 ≈ 111 km）

        if distance < min_distance:
            min_distance = distance
            nearest_team = {
                'team_id': team_id,
                'distance_km': round(distance, 1),
                'eta_minutes': int((distance / 50) * 60)  # 假設平均速度 50 km/h
            }

    return nearest_team

def lambda_handler(event, context):
    """
    爆管確認後自動執行：
    1. 推算爆管位置
    2. 派遣最近維修團隊
    3. 通知相關單位
    4. 啟動應急預案
    """

    sensor_id = event['detail']['sensor_id']
    pressure = event['detail']['pressure_bar']
    flow = event['detail']['flow_m3h']

    # 步驟 1：推算位置
    burst_location = calculate_burst_location(sensor_id, pressure, flow)

    # 步驟 2：派遣維修
    repair_team = find_nearest_repair_team(burst_location)

    # 步驟 3：通知 SNS
    alert_message = {
        'alert_type': 'GAS_PIPE_BURST_CONFIRMED',
        'timestamp': datetime.utcnow().isoformat() + 'Z',
        'burst_location': burst_location,
        'repair_team_dispatched': repair_team,
        'actions': [
            'CLOSE_DOWNSTREAM_VALVE',
            'EVACUATE_NEARBY_AREA',
            'CONTACT_FIRE_DEPARTMENT',
            'DISPATCH_REPAIR_TEAM',
            'NOTIFY_CITY_GOVERNMENT'
        ]
    }

    # 通知高管（30 秒內）
    sns.publish(
        TopicArn='arn:aws:sns:tw-south-1:123456789012:gas-burst-alert-critical',
        Subject='🚨 [CRITICAL] 瓦斯管線爆管 - 立即行動',
        Message=json.dumps(alert_message, indent=2, ensure_ascii=False)
    )

    # 存儲事件記錄
    table = dynamodb.Table('GasBurstEvents')
    table.put_item(
        Item={
            'sensor_id': sensor_id,
            'timestamp': int(datetime.utcnow().timestamp()),
            'burst_location': burst_location,
            'repair_team': repair_team,
            'status': 'DISPATCHED'
        }
    )

    # 步驟 4：啟動應急預案（另一個 Lambda）
    lambda_client.invoke(
        FunctionName='EmergencyResponsePlan',
        InvocationType='Event',
        Payload=json.dumps({
            'burst_location': burst_location,
            'repair_team': repair_team
        })
    )

    return {
        'statusCode': 200,
        'burst_location': burst_location,
        'repair_team_dispatched': repair_team,
        'message': '爆管已確認，維修團隊已派遣'
    }

【IoT Events Input】
{
  "inputName": "GaslineInput",
  "inputDescription": "LNG 管線即時數據輸入",
  "inputDefinition": {
    "attributes": [
      {"jsonPath": "$.sensor_id"},
      {"jsonPath": "$.pressure_bar"},
      {"jsonPath": "$.flow_m3h"},
      {"jsonPath": "$.temperature_c"},
      {"jsonPath": "$.timestamp"}
    ]
  }
}

【IoT Events Actions 配置】
SNS Action：
  - Topic: gas-burst-alert-critical
  - Priority: IMMEDIATE (30 秒內必須通知)
  - Recipients: 董事長、營運長、安全主管、消防局

Lambda Action：
  - Function: BurstLocationCalculator
  - Timeout: 10 秒（快速推算位置）

DynamoDB Action：
  - Table: GasBurstEvents
  - TTL: 7 年（法定保存期限）

CloudWatch Logs Action：
  - Log Group: /aws/iotevents/gas-monitoring
  - 完整審計軌跡

【告警優先級】
1. CRITICAL: 爆管確認 → 30 秒內通知所有人員
2. WARNING: 異常壓力 → 5 分鐘內檢查
3. INFO: 壓力波動正常 → 記錄日誌