Python：UPW 品質即時監控

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

class UPWMonitor:
    """
    超纯水監控系統
    監控對象：
    - 電導率（Conductivity）0.055~18.2 MΩ·cm
    - 總有机碳（TOC）0~10 ppb
    - 顆粒度（Particle Count）0.5~100 µm
    - 流速（Flow Rate）0~500 L/min
    """

    def __init__(self, conductivity_port='/dev/ttyUSB0', toc_port='/dev/ttyUSB1'):
        self.conductivity_serial = serial.Serial(conductivity_port, 9600, timeout=2.0)
        self.toc_serial = serial.Serial(toc_port, 9600, timeout=2.0)

        # 歷史數據（用於趨勢分析）
        self.conductivity_history = deque(maxlen=600)  # 10分鐘（1秒采样）
        self.toc_history = deque(maxlen=600)
        self.flow_history = deque(maxlen=600)

        # 告警閾值
        self.conductivity_warning_threshold = 0.5  # µS/cm
        self.conductivity_critical_threshold = 1.0  # µS/cm
        self.toc_threshold = 5.0  # ppb
        self.flow_drop_threshold = 20  # % 流速下降

    def read_conductivity(self):
        """
        讀取電導率（Modbus RTU）
        寄存器：0x0000，數據格式：IEEE 754浮點
        返回值單位：µS/cm
        """
        try:
            self.conductivity_serial.write(
                bytes([0x01, 0x03, 0x00, 0x00, 0x00, 0x02, 0xC4, 0x0B])
            )
            response = self.conductivity_serial.read(7)

            if len(response) < 7:
                return {'value': None, 'status': 'TIMEOUT'}

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

            self.conductivity_history.append(conductivity_uS)

            # 即時判斷
            if conductivity_uS > self.conductivity_critical_threshold:
                return {
                    'value': round(conductivity_uS, 3),
                    'status': 'CRITICAL',
                    'reason': f'電導率 {conductivity_uS:.3f} µS/cm 超過閾值 {self.conductivity_critical_threshold}'
                }
            elif conductivity_uS > self.conductivity_warning_threshold:
                return {
                    'value': round(conductivity_uS, 3),
                    'status': 'WARNING',
                    'reason': f'電導率 {conductivity_uS:.3f} µS/cm 接近閾值'
                }
            else:
                return {
                    'value': round(conductivity_uS, 3),
                    'status': 'OK'
                }

        except Exception as e:
            return {'value': None, 'status': 'SENSOR_ERROR', 'error': str(e)}

    def read_toc(self):
        """
        讀取總有機碳（TOC）
        範圍：0~10 ppb，精度：±0.5 ppb
        """
        try:
            self.toc_serial.write(
                bytes([0x02, 0x03, 0x00, 0x10, 0x00, 0x02, 0x74, 0x1A])
            )
            response = self.toc_serial.read(7)

            if len(response) < 7:
                return {'value': None, 'status': 'TIMEOUT'}

            toc_ppb = struct.unpack('>f', response[3:7])[0]
            self.toc_history.append(toc_ppb)

            if toc_ppb > self.toc_threshold:
                return {
                    'value': round(toc_ppb, 2),
                    'status': 'CRITICAL',
                    'reason': f'TOC {toc_ppb:.2f} ppb 超標'
                }
            else:
                return {
                    'value': round(toc_ppb, 2),
                    'status': 'OK'
                }

        except Exception as e:
            return {'value': None, 'status': 'SENSOR_ERROR'}

    def analyze_trends(self):
        """
        分析歷史趨勢，預測故障
        """
        if len(self.conductivity_history) < 60:  # 至少60秒數據
            return {'prediction': 'INSUFFICIENT_DATA'}

        cond_arr = np.array(list(self.conductivity_history), dtype=np.float32)

        # 計算傾斜度
        slope = np.polyfit(range(len(cond_arr)), cond_arr, 1)[0]

        # 標準差
        std_dev = float(np.std(cond_arr))

        # 預測
        if slope > 0.002:  # 電導率上升趨勢
            prediction = 'DEGRADATION_DETECTED'
            recommendation = '減速 UPW 流量或考慮更換濾芯'
        elif std_dev > 0.15:  # 高波動
            prediction = 'INSTABILITY'
            recommendation = '檢查 UPW 系統壓力和流速'
        else:
            prediction = 'HEALTHY'
            recommendation = 'CONTINUE'

        return {
            'prediction': prediction,
            'slope': round(slope, 6),
            'std_dev': round(std_dev, 4),
            'recommendation': recommendation
        }

    def get_upw_status(self):
        """
        綜合 UPW 狀態報告
        """
        conductivity_data = self.read_conductivity()
        toc_data = self.read_toc()
        trend_analysis = self.analyze_trends()

        # 綜合評級
        if conductivity_data['status'] == 'CRITICAL' or toc_data['status'] == 'CRITICAL':
            overall_status = 'CRITICAL'
            action = 'STOP_UPW_SUPPLY'
        elif conductivity_data['status'] == 'WARNING' or trend_analysis['prediction'] == 'DEGRADATION_DETECTED':
            overall_status = 'WARNING'
            action = 'ALERT_MAINTENANCE'
        else:
            overall_status = 'OK'
            action = 'CONTINUE'

        return {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'overall_status': overall_status,
            'conductivity_uS': conductivity_data,
            'toc_ppb': toc_data,
            'trend_analysis': trend_analysis,
            'immediate_action': action,
            'estimated_upw_lifespan': self._estimate_upw_lifespan()
        }

    def _estimate_upw_lifespan(self):
        """估計 UPW 系統剩餘壽命"""
        if len(self.conductivity_history) < 3600:  # 至少1小時數據
            return 'UNKNOWN'

        cond_arr = np.array(list(self.conductivity_history), dtype=np.float32)
        slope = np.polyfit(range(len(cond_arr)), cond_arr, 1)[0]

        if slope <= 0:
            return 'INDEFINITE'

        # 從當前值到臨界值需要多久
        current = cond_arr[-1]
        critical = self.conductivity_critical_threshold

        if current >= critical:
            return 'EXPIRED_NOW'

        hours_remaining = (critical - current) / (slope * 3600)  # 轉換為小時

        return {
            'hours_remaining': round(hours_remaining, 1),
            'estimated_expiry': datetime.utcnow().isoformat() + 'Z'
        }

def lambda_handler(event, context):
    """邊界 Lambda 入口點"""
    monitor = UPWMonitor()
    upw_status = monitor.get_upw_status()

    # 記錄到本地 DynamoDB
    return {
        'statusCode': 200,
        'upw_status': upw_status
    }

Python：化學槽健康度評分（0~100）

import json
import numpy as np
import tensorflow as tf
import time
from datetime import datetime, timedelta
from collections import deque

MODEL_PATH = "/greengrass/ml/chemical_tank_predictive_model.tflite"
interpreter = tf.lite.Interpreter(model_path=MODEL_PATH)
interpreter.allocate_tensors()

input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

class ChemicalTankPredictor:
    """
    化學槽預防性維護預測
    監控壓力、溫度、流速變化，預測故障前置時間
    """

    def __init__(self, tank_id='TANK_001'):
        self.tank_id = tank_id

        # 歷史數據（24小時）
        self.pressure_history = deque(maxlen=1440)  # 1分鐘采样，24小時
        self.temp_history = deque(maxlen=1440)
        self.flow_history = deque(maxlen=1440)
        self.turbidity_history = deque(maxlen=1440)

        # 化學槽壽命統計
        self.tank_lifespan_expected = 180  # 天（3~6個月）
        self.tank_installation_date = datetime.utcnow()

        # 故障預測閾值
        self.pressure_variance_threshold = 0.15  # 15% 壓力波動
        self.temp_drift_threshold = 2.0  # ±2°C 溫度漂移
        self.flow_drop_threshold = 20  # % 流速下降

    def extract_predictive_features(self, pressure_values, temp_values, flow_values, turbidity_values):
        """
        提取 12 個預測特徵
        """
        if len(pressure_values) < 60:  # 至少1小時數據
            return None

        p_arr = np.array(list(pressure_values[-60:]), dtype=np.float32)
        t_arr = np.array(list(temp_values[-60:]), dtype=np.float32)
        f_arr = np.array(list(flow_values[-60:]), dtype=np.float32)
        turb_arr = np.array(list(turbidity_values[-60:]), dtype=np.float32)

        # 壓力特徵 (4)
        p_mean = float(np.mean(p_arr))
        p_std = float(np.std(p_arr))
        p_slope = float(np.polyfit(range(len(p_arr)), p_arr, 1)[0])
        p_variance_ratio = p_std / (abs(p_mean) + 1e-6)

        # 溫度特徵 (3)
        t_mean = float(np.mean(t_arr))
        t_std = float(np.std(t_arr))
        t_slope = float(np.polyfit(range(len(t_arr)), t_arr, 1)[0])

        # 流速特徵 (3)
        f_mean = float(np.mean(f_arr))
        f_drop_ratio = (f_arr[0] - f_arr[-1]) / f_arr[0] if f_arr[0] > 0 else 0
        f_stability = float(np.std(f_arr) / (np.mean(f_arr) + 1e-6))

        # 濁度特徵 (2)
        turb_mean = float(np.mean(turb_arr))
        turb_trend = float(np.polyfit(range(len(turb_arr)), turb_arr, 1)[0])

        features = np.array([
            p_mean, p_std, p_slope, p_variance_ratio,
            t_mean, t_std, t_slope,
            f_mean, f_drop_ratio, f_stability,
            turb_mean, turb_trend
        ], dtype=np.float32)

        return features

    def predict_tank_failure_time(self, pressure, temp, flow_rate, turbidity, tank_age_days):
        """
        預測化學槽故障前置時間（小時）
        """
        self.pressure_history.append(pressure)
        self.temp_history.append(temp)
        self.flow_history.append(flow_rate)
        self.turbidity_history.append(turbidity)

        # 即時範圍檢查
        if pressure < 1.5 or pressure > 8.0:
            return {
                'health_score': 0,
                'failure_risk': 'CRITICAL',
                'hours_to_failure': 0,
                'reason': '壓力異常',
                'immediate_action': 'STOP_PROCESS'
            }

        if len(self.pressure_history) < 60:
            return {
                'health_score': 50,
                'failure_risk': 'UNKNOWN',
                'hours_to_failure': -1,
                'reason': '數據不足'
            }

        try:
            features = self.extract_predictive_features(
                self.pressure_history, self.temp_history,
                self.flow_history, self.turbidity_history
            )

            if features is None:
                return {'health_score': 50, 'failure_risk': 'UNKNOWN'}

            # 正規化特徵
            features_normalized = features / (np.linalg.norm(features) + 1e-6)
            features_reshaped = features_normalized.reshape((1, -1))

            # TF Lite 推論
            interpreter.set_tensor(input_details[0]['index'], features_reshaped)
            interpreter.invoke()

            output_data = interpreter.get_tensor(output_details[0]['index'])
            failure_probability = float(output_data[0][0])

            # 計算健康評分和故障預測時間
            health_score = int(100 * (1 - failure_probability))

            if failure_probability > 0.8:
                failure_risk = 'CRITICAL'
                hours_to_failure = 12  # 12小時內需要更換
                maintenance_type = 'EMERGENCY'
            elif failure_probability > 0.6:
                failure_risk = 'HIGH'
                hours_to_failure = 48  # 48小時內計畫更換
                maintenance_type = 'URGENT'
            elif failure_probability > 0.4:
                failure_risk = 'MEDIUM'
                hours_to_failure = 96  # 96小時內計畫更換
                maintenance_type = 'PLANNED'
            else:
                failure_risk = 'LOW'
                hours_to_failure = -1  # 暫無計畫
                maintenance_type = 'MONITOR'

            # 結合槽齡預測
            age_factor = tank_age_days / self.tank_lifespan_expected
            if age_factor > 0.9:
                health_score = max(0, health_score - 20)  # 老化扣分
                failure_risk = 'HIGH'

            return {
                'tank_id': self.tank_id,
                'health_score': health_score,
                'failure_risk': failure_risk,
                'failure_probability': round(failure_probability, 4),
                'hours_to_failure': hours_to_failure,
                'maintenance_type': maintenance_type,
                'tank_age_days': tank_age_days,
                'tank_remaining_lifespan_percent': round(100 * (1 - age_factor), 1),
                'next_maintenance_window': self._calculate_maintenance_window(hours_to_failure)
            }

        except Exception as e:
            return {
                'health_score': 50,
                'failure_risk': 'INFERENCE_ERROR',
                'error': str(e)
            }

    def _calculate_maintenance_window(self, hours_to_failure):
        """計算最佳維護窗口"""
        if hours_to_failure <= 0:
            return None

        now = datetime.utcnow()
        maintenance_time = now + timedelta(hours=hours_to_failure - 4)  # 提前4小時通知

        return {
            'recommended_time': maintenance_time.isoformat() + 'Z',
            'duration_minutes': 120  # 更換通常需要2小時
        }

def lambda_handler(event, context):
    """邊界 Lambda 入口點"""
    predictor = ChemicalTankPredictor(tank_id=event.get('tank_id', 'TANK_001'))

    pressure = event.get('pressure_bar')
    temp = event.get('temperature_c')
    flow_rate = event.get('flow_lpm')
    turbidity = event.get('turbidity_ntu')
    tank_age_days = event.get('tank_age_days', 0)

    prediction = predictor.predict_tank_failure_time(
        pressure, temp, flow_rate, turbidity, tank_age_days
    )

    return {
        'statusCode': 200,
        'predictive_maintenance': prediction
    }

Python：冷卻水溫度差異分析

import json
import numpy as np
from datetime import datetime
from collections import deque

class ChilledWaterMonitor:
    """
    冷卻水系統監控
    監控進/出水溫度、流速、壓差
    預測洩漏、結垢、泵故障
    """

    def __init__(self, equipment_id='EQUIP_001'):
        self.equipment_id = equipment_id
        self.inlet_temp_history = deque(maxlen=600)   # 10分鐘
        self.outlet_temp_history = deque(maxlen=600)
        self.flow_history = deque(maxlen=600)
        self.pressure_diff_history = deque(maxlen=600)

        # 設備參數（根據實際設備設置）
        self.nominal_flow_lpm = 100  # 標稱流速
        self.nominal_delta_t = 5.0   # 標稱溫度差（°C）
        self.nominal_pressure_drop = 2.5  # 標稱壓差（bar）

    def monitor_coolant_system(self, inlet_temp, outlet_temp, flow_rate, pressure_drop):
        """
        監控冷卻系統健康度
        """
        self.inlet_temp_history.append(inlet_temp)
        self.outlet_temp_history.append(outlet_temp)
        self.flow_history.append(flow_rate)
        self.pressure_diff_history.append(pressure_drop)

        delta_t = outlet_temp - inlet_temp
        anomalies = []

        # 1. 溫度異常檢測
        if delta_t < 0.5:
            anomalies.append({
                'type': 'LOW_DELTA_T',
                'severity': 'HIGH',
                'description': '溫度差異過小，可能流速過快或設備未負載'
            })
        elif delta_t > self.nominal_delta_t + 3:
            anomalies.append({
                'type': 'HIGH_DELTA_T',
                'severity': 'HIGH',
                'description': '溫度差異過大，可能冷却能力不足'
            })

        # 2. 流速異常檢測
        flow_drop = (self.nominal_flow_lpm - flow_rate) / self.nominal_flow_lpm * 100

        if flow_drop > 20:
            anomalies.append({
                'type': 'FLOW_DROP',
                'severity': 'CRITICAL',
                'description': f'流速下降 {flow_drop:.1f}%，可能洩漏或泵故障'
            })

        # 3. 壓差異常檢測
        pressure_diff_variance = abs(pressure_drop - self.nominal_pressure_drop) / self.nominal_pressure_drop

        if pressure_diff_variance > 0.3:
            anomalies.append({
                'type': 'PRESSURE_ANOMALY',
                'severity': 'HIGH',
                'description': f'壓差異常 {pressure_diff_variance*100:.1f}%'
            })

        # 4. 趨勢分析（結垢/劣化）
        if len(self.pressure_diff_history) > 60:
            p_arr = np.array(list(self.pressure_diff_history), dtype=np.float32)
            slope = np.polyfit(range(len(p_arr)), p_arr, 1)[0]

            if slope > 0.01:  # 壓差上升趨勢
                anomalies.append({
                    'type': 'FOULING_DETECTED',
                    'severity': 'MEDIUM',
                    'description': '檢測到結垢趨勢，建議定期清洗'
                })

        # 綜合評分
        if any(a['severity'] == 'CRITICAL' for a in anomalies):
            system_status = 'CRITICAL'
            action = 'STOP_EQUIPMENT'
        elif any(a['severity'] == 'HIGH' for a in anomalies):
            system_status = 'WARNING'
            action = 'ALERT_MAINTENANCE'
        else:
            system_status = 'OK'
            action = 'CONTINUE'

        return {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'equipment_id': self.equipment_id,
            'system_status': system_status,
            'immediate_action': action,
            'temperatures': {
                'inlet_c': round(inlet_temp, 2),
                'outlet_c': round(outlet_temp, 2),
                'delta_t': round(delta_t, 2)
            },
            'flow_metrics': {
                'current_lpm': round(flow_rate, 1),
                'nominal_lpm': self.nominal_flow_lpm,
                'drop_percent': round(flow_drop, 1)
            },
            'pressure_metrics': {
                'current_bar': round(pressure_drop, 2),
                'nominal_bar': self.nominal_pressure_drop
            },
            'anomalies': anomalies
        }

def lambda_handler(event, context):
    """邊界 Lambda 入口點"""
    monitor = ChilledWaterMonitor(equipment_id=event.get('equipment_id', 'EQUIP_001'))

    inlet_temp = event.get('inlet_temperature_c')
    outlet_temp = event.get('outlet_temperature_c')
    flow_rate = event.get('flow_rate_lpm')
    pressure_drop = event.get('pressure_drop_bar')

    result = monitor.monitor_coolant_system(inlet_temp, outlet_temp, flow_rate, pressure_drop)

    return {
        'statusCode': 200,
        'coolant_system': result
    }

SELECT
  timestamp() as alert_timestamp,
  equipment_id,
  health_score,
  failure_risk,
  hours_to_failure,
  maintenance_type,
  upw_status.overall_status as upw_status,
  coolant_system.system_status as coolant_status
FROM 'semiconductor/fab/equipment/+/telemetry'
WHERE health_score < 60 OR upw_status.overall_status = 'CRITICAL' OR coolant_system.system_status = 'CRITICAL'

ACTION:
  DynamoDB:
    RoleArn: arn:aws:iam::123456789012:role/iot-maintenance-role
    TableName: PreventiveMaintenanceAlerts
    HashKeyValue: ${equipment_id}
    RangeKeyValue: ${timestamp()}
    Item:
      equipment_id
      alert_timestamp: timestamp()
      health_score
      failure_risk
      hours_to_failure
      maintenance_type
      ttl: ${timestamp() + 7776000}

ACTION:
  Sns:
    RoleArn: arn:aws:iam::123456789012:role/iot-sns-role
    TargetArn: arn:aws:sns:us-west-1:123456789012:fab-maintenance-alerts
    MessageFormat: JSON

import json
import boto3
from datetime import datetime
import numpy as np

dynamodb = boto3.resource('dynamodb')
sns = boto3.client('sns')
cloudwatch = boto3.client('cloudwatch')

def lambda_handler(event, context):
    """
    綜合評估：UPW + 化學槽 + 水冷系統
    生成維護優先級和成本預估
    """

    equipment_id = event.get('equipment_id')
    upw_status = event.get('upw_status')
    tank_prediction = event.get('predictive_maintenance')
    coolant_status = event.get('coolant_system')

    # 計算綜合風險評分（0~100）
    risk_scores = []

    if upw_status.get('overall_status') == 'CRITICAL':
        risk_scores.append(100)
    elif upw_status.get('overall_status') == 'WARNING':
        risk_scores.append(70)
    else:
        risk_scores.append(20)

    risk_scores.append(tank_prediction.get('health_score', 50))

    if coolant_status.get('system_status') == 'CRITICAL':
        risk_scores.append(100)
    elif coolant_status.get('system_status') == 'WARNING':
        risk_scores.append(70)
    else:
        risk_scores.append(20)

    overall_risk_score = int(np.mean(risk_scores))

    # 生成維護計劃
    maintenance_plan = generate_maintenance_plan(
        equipment_id,
        overall_risk_score,
        tank_prediction,
        upw_status,
        coolant_status
    )

    # 發送通知
    if overall_risk_score > 75:
        send_alert(equipment_id, maintenance_plan, 'URGENT')
    elif overall_risk_score > 50:
        send_alert(equipment_id, maintenance_plan, 'PLANNED')

    # 記錄到 CloudWatch
    cloudwatch.put_metric_data(
        Namespace='SemiconductorFab',
        MetricData=[{
            'MetricName': 'EquipmentRiskScore',
            'Value': overall_risk_score,
            'Unit': 'None',
            'Dimensions': [
                {'Name': 'EquipmentID', 'Value': equipment_id}
            ]
        }]
    )

    return {
        'statusCode': 200,
        'equipment_id': equipment_id,
        'overall_risk_score': overall_risk_score,
        'maintenance_plan': maintenance_plan
    }

def generate_maintenance_plan(equipment_id, risk_score, tank_pred, upw, coolant):
    """生成詳細維護計畫"""
    tasks = []

    # UPW 維護任務
    if upw.get('overall_status') in ['CRITICAL', 'WARNING']:
        tasks.append({
            'priority': 'CRITICAL' if upw.get('overall_status') == 'CRITICAL' else 'HIGH',
            'task': 'UPW_SYSTEM_INSPECTION',
            'description': upw.get('immediate_action'),
            'estimated_duration_minutes': 30
        })

    # 化學槽維護任務
    if tank_pred.get('failure_risk') in ['CRITICAL', 'HIGH']:
        tasks.append({
            'priority': tank_pred.get('maintenance_type').upper(),
            'task': 'CHEMICAL_TANK_REPLACEMENT',
            'description': f'槽體健康度 {tank_pred.get("health_score")}%，故障預測 {tank_pred.get("hours_to_failure")} 小時內',
            'estimated_duration_minutes': 120,
            'next_window': tank_pred.get('next_maintenance_window')
        })

    # 冷卻水維護任務
    if coolant.get('system_status') in ['CRITICAL', 'WARNING']:
        for anomaly in coolant.get('anomalies', []):
            tasks.append({
                'priority': 'CRITICAL' if anomaly['severity'] == 'CRITICAL' else 'HIGH',
                'task': anomaly['type'],
                'description': anomaly['description'],
                'estimated_duration_minutes': 60 if anomaly['type'] == 'FOULING_DETECTED' else 120
            })

    # 估計維修成本
    total_downtime = sum(t['estimated_duration_minutes'] for t in tasks)
    cost_per_hour = 500000  # USD/小時
    estimated_cost = (total_downtime / 60) * cost_per_hour

    return {
        'timestamp': datetime.utcnow().isoformat() + 'Z',
        'equipment_id': equipment_id,
        'tasks': tasks,
        'total_tasks': len(tasks),
        'estimated_downtime_minutes': total_downtime,
        'estimated_cost_usd': round(estimated_cost),
        'urgency_level': 'CRITICAL' if risk_score > 75 else ('URGENT' if risk_score > 50 else 'PLANNED')
    }

def send_alert(equipment_id, plan, priority):
    """發送 SNS 告警"""
    message = {
        'equipment_id': equipment_id,
        'maintenance_plan': plan,
        'priority': priority
    }

    sns.publish(
        TopicArn='arn:aws:sns:us-west-1:123456789012:fab-maintenance-alerts',
        Subject=f'[{priority}] {equipment_id} 需要預防性維護',
        Message=json.dumps(message, indent=2)
    )