การบำรุงรักษากังหันลม (Wind Turbine Maintenance) ในยุคปัจจุบันต้องอาศัย AI วิเคราะห์ข้อมูลจากเซ็นเซอร์หลายร้อยตัวพร้อมกัน ไม่ว่าจะเป็นการตรวจจับความผิดปกติของระบบเกียร์ การวิเคราะห์แบบจำลองความเสียหาย หรือการสร้างรายงานอัตโนมัติ บทความนี้จะพาคุณสร้างระบบ Wind Power O&M (Operations & Maintenance) AI Platform ที่ใช้ HolySheep AI เป็น Backbone โดยเน้นการประมวลผล传感数据归因 (Sensor Data Attribution) และการใช้ Claude สำหรับ故障复盘 (Fault Review) พร้อมทั้งสอนการตั้งค่า SLA Switching สำหรับโมเดลต่างๆ

สถาปัตยกรรมระบบ Wind Turbine O&M AI

ระบบที่เราจะสร้างประกอบด้วย 3 ชั้นหลัก:

import requests
import json
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class SensorReading:
    turbine_id: str
    timestamp: datetime
    vibration_x: float
    vibration_y: float
    vibration_z: float
    temperature_bearing: float
    temperature_gearbox: float
    rpm: float
    power_output: float
    wind_speed: float

class HolySheepWindOM:
    """
    HolySheep AI Client สำหรับระบบ Wind Turbine O&M
    Base URL: https://api.holysheep.ai/v1
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def analyze_sensor_anomaly(self, sensor_data: Dict) -> Dict:
        """
        วิเคราะห์ความผิดปกติของเซ็นเซอร์ด้วย Claude Sonnet 4.5
        ใช้ prompt สำหรับ wind turbine fault diagnosis
        """
        prompt = f"""คุณคือวิศวกรผู้เชี่ยวชาญด้านกังหันลม
ข้อมูลเซ็นเซอร์จากกังหันลม {sensor_data.get('turbine_id')}:
- ความสั่นสะเทือน X: {sensor_data.get('vibration_x')} mm/s
- ความสั่นสะเทือน Y: {sensor_data.get('vibration_y')} mm/s  
- ความสั่นสะเทือน Z: {sensor_data.get('vibration_z')} mm/s
- อุณหภูมิตลับลูกปืน: {sensor_data.get('temperature_bearing')} °C
- อุณหภูมิเกียร์บ็อกซ์: {sensor_data.get('temperature_gearbox')} °C
- ความเร็วรอบ: {sensor_data.get('rpm')} RPM
- กำลังการผลิต: {sensor_data.get('power_output')} kW
- ความเร็วลม: {sensor_data.get('wind_speed')} m/s

วิเคราะห์และให้:
1. Root Cause Analysis (สาเหตุหลัก)
2. Severity Level (ระดับความรุนแรง: Critical/High/Medium/Low)
3. Recommended Actions (การดำเนินการที่แนะนำ)
4. Maintenance Priority Score (คะแนนความสำคัญ 1-100)
"""
        
        payload = {
            "model": "claude-sonnet-4.5",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 1500,
            "temperature": 0.3
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"HolySheep API Error: {response.status_code}")


ตัวอย่างการใช้งาน

api_client = HolySheepWindOM(api_key="YOUR_HOLYSHEEP_API_KEY")

sensor_data = {
    "turbine_id": "WT-2024-CN-0042",
    "vibration_x": 4.8,
    "vibration_y": 3.2,
    "vibration_z": 6.1,
    "temperature_bearing": 87.5,
    "temperature_gearbox": 92.3,
    "rpm": 14.5,
    "power_output": 2850,
    "wind_speed": 11.2
}

result = api_client.analyze_sensor_anomaly(sensor_data)
print(f"Fault Analysis: {result['choices'][0]['message']['content']}")

การตั้งค่า SLA Switching สำหรับ Multi-Model Orchestration

ในระบบ Production จริง การใช้โมเดล AI เพียงตัวเดียวไม่เพียงพอ คุณต้องมีกลยุทธ์การสลับโมเดลตาม:

from enum import Enum
from typing import Callable
import time

class ModelTier(Enum):
    PREMIUM = "claude-sonnet-4.5"       # $15/MTok - วิเคราะห์ลึก
    STANDARD = "gpt-4.1"                # $8/MTok - งานทั่วไป
    FAST = "gemini-2.5-flash"           # $2.50/MTok - ตอบเร็ว
    BUDGET = "deepseek-v3.2"            # $0.42/MTok - งานง่าย

class SLAConfig:
    """กำหนดค่า SLA สำหรับแต่ละประเภทงาน"""
    
    def __init__(self):
        self.sla_tiers = {
            "critical_fault": {
                "model": ModelTier.PREMIUM,
                "max_latency_ms": 5000,
                "min_accuracy": 0.95,
                "fallback": ModelTier.STANDARD
            },
            "routine_check": {
                "model": ModelTier.BUDGET,
                "max_latency_ms": 500,
                "min_accuracy": 0.85,
                "fallback": ModelTier.FAST
            },
            "trend_analysis": {
                "model": ModelTier.STANDARD,
                "max_latency_ms": 2000,
                "min_accuracy": 0.90,
                "fallback": ModelTier.FAST
            },
            "emergency_alert": {
                "model": ModelTier.FAST,
                "max_latency_ms": 300,
                "min_accuracy": 0.80,
                "fallback": ModelTier.BUDGET
            }
        }
    
    def get_model_for_task(self, task_type: str) -> str:
        config = self.sla_tiers.get(task_type, self.sla_tiers["routine_check"])
        return config["model"].value

class SLASwitchingOrchestrator:
    """
    Orchestrator สำหรับจัดการการสลับโมเดลตาม SLA
    รองรับ Fallback chain และ Performance monitoring
    """
    
    def __init__(self, api_client: HolySheepWindOM, sla_config: SLAConfig):
        self.client = api_client
        self.sla_config = sla_config
        self.usage_stats = {}
        self.latency_stats = {}
    
    def execute_with_sla(
        self, 
        task_type: str, 
        prompt: str,
        enable_fallback: bool = True
    ) -> Dict:
        """Execute task with SLA-aware model selection"""
        
        start_time = time.time()
        model = self.sla_config.get_model_for_task(task_type)
        sla = self.sla_config.sla_tiers.get(task_type, {})
        
        try:
            result = self._call_model(model, prompt)
            
            # Track stats
            latency = (time.time() - start_time) * 1000
            self._track_stats(model, latency, success=True)
            
            # Check SLA compliance
            if latency > sla.get("max_latency_ms", 1000):
                print(f"⚠️ Latency SLA breached: {latency:.0f}ms > {sla['max_latency_ms']}ms")
            
            return {
                "result": result,
                "model_used": model,
                "latency_ms": latency,
                "sla_compliant": latency <= sla.get("max_latency_ms", 1000)
            }
            
        except Exception as e:
            if enable_fallback and sla.get("fallback"):
                print(f"🔄 Falling back to {sla['fallback'].value}")
                return self._fallback_execution(sla["fallback"], prompt)
            raise
    
    def _call_model(self, model: str, prompt: str) -> str:
        """เรียก HolySheep API ด้วยโมเดลที่กำหนด"""
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 1000,
            "temperature": 0.4
        }
        
        response = requests.post(
            f"{self.client.base_url}/chat/completions",
            headers=self.client.headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code == 200:
            return response.json()["choices"][0]["message"]["content"]
        else:
            raise Exception(f"Model call failed: {response.status_code}")
    
    def _track_stats(self, model: str, latency: float, success: bool):
        """ติดตามสถิติการใช้งาน"""
        if model not in self.usage_stats:
            self.usage_stats[model] = {"calls": 0, "failures": 0, "total_latency": 0}
        
        self.usage_stats[model]["calls"] += 1
        self.usage_stats[model]["total_latency"] += latency
        if not success:
            self.usage_stats[model]["failures"] += 1
    
    def get_usage_report(self) -> Dict:
        """สร้างรายงานการใช้งานแต่ละโมเดล"""
        report = {}
        for model, stats in self.usage_stats.items():
            report[model] = {
                "total_calls": stats["calls"],
                "failure_rate": stats["failures"] / stats["calls"] if stats["calls"] > 0 else 0,
                "avg_latency_ms": stats["total_latency"] / stats["calls"] if stats["calls"] > 0 else 0
            }
        return report


ตัวอย่างการใช้งาน

sla_config = SLAConfig()
orchestrator = SLASwitchingOrchestrator(api_client, sla_config)


งาน Critical - ใช้ Claude Sonnet 4.5

result1 = orchestrator.execute_with_sla(
    "critical_fault",
    "วิเคราะห์ความผิดปกติ: ความสั่นสะเทือน Z สูงผิดปกติ 6.1 mm/s"
)


งาน Routine - ใช้ DeepSeek V3.2

result2 = orchestrator.execute_with_sla(
    "routine_check", 
    "ตรวจสอบสถานะทั่วไปของกังหันลม WT-001"
)

print(orchestrator.get_usage_report())

传感数据归因: ระบบวิเคราะห์ข้อมูลเซ็นเซอร์แบบ Multi-Source

การ归因 (Attribution) ของข้อมูลเซ็นเซอร์เป็นหัวใจสำคัญในการเข้าใจว่าปัญหาเกิดจากสาเหตุใด ในระบบ Wind Turbine จะมีเซ็นเซอร์หลายประเภทที่ต้องรวมข้อมูลเข้าด้วยกัน:

import pandas as pd
from collections import defaultdict
from datetime import timedelta

class SensorAttributionEngine:
    """
    ระบบวิเคราะห์การ归因 (Attribution) ของข้อมูลเซ็นเซอร์
    ใช้ AI ระบุว่าความผิดปกติเกิดจากเซ็นเซอร์ตัวใดหรือ Component ใด
    """
    
    def __init__(self, api_client: HolySheepWindOM):
        self.client = api_client
        self.correlation_threshold = 0.7
    
    def correlate_sensors(self, sensor_df: pd.DataFrame) -> Dict:
        """
        หาความสัมพันธ์ระหว่างเซ็นเซอร์หลายตัว
        ส่งข้อมูลให้ Claude วิเคราะห์ Root Cause Chain
        """
        
        # คำนวณ Correlation Matrix
        numeric_cols = sensor_df.select_dtypes(include=['float64', 'int64']).columns
        correlation_matrix = sensor_df[numeric_cols].corr()
        
        # หา Sensor ที่มีความสัมพันธ์สูง
        high_correlation_pairs = []
        for i, col1 in enumerate(correlation_matrix.columns):
            for j, col2 in enumerate(correlation_matrix.columns):
                if i < j:
                    corr = abs(correlation_matrix.loc[col1, col2])
                    if corr > self.correlation_threshold:
                        high_correlation_pairs.append({
                            "sensor_1": col1,
                            "sensor_2": col2,
                            "correlation": corr
                        })
        
        return {
            "high_correlation_pairs": high_correlation_pairs,
            "correlation_matrix": correlation_matrix.to_dict()
        }
    
    def generate_attribution_report(
        self, 
        turbine_id: str, 
        anomaly_data: Dict,
        historical_data: pd.DataFrame
    ) -> str:
        """
        สร้างรายงาน Attribution ด้วย Claude Sonnet 4.5
        ระบุว่า Anomaly เกิดจาก Component ใดเป็นหลัก
        """
        
        # รวบรวมข้อมูลเซ็นเซอร์ที่ผิดปกติ
        anomaly_sensors = anomaly_data.get("anomalous_sensors", [])
        anomaly_timeline = anomaly_data.get("timeline", [])
        
        # สร้าง Context สำหรับ AI
        context = f"""รายงานการวิเคราะห์归因 (Attribution) สำหรับกังหันลม {turbine_id}

ข้อมูลเซ็นเซอร์ที่แสดงความผิดปกติ:
{self._format_sensor_list(anomaly_sensors)}

ข้อมูลประวัติย้อนหลัง (สถิติ):
- ค่าเฉลี่ย: {historical_data.mean().to_dict()}
- ค่าเบี่ยงเบนมาตรฐาน: {historical_data.std().to_dict()}
- ค่าสูงสุด: {historical_data.max().to_dict()}

เซ็นเซอร์หลักที่ควรพิจารณา:
1. Vibration Sensors (ความสั่นสะเทือน) - บ่งบอกปัญหา mechanical
2. Temperature Sensors - บ่งบอกปัญหาความร้อน/การเสียดสี
3. RPM Sensors - บ่งบอกปัญหาเกี่ยวกับ Speed controller
4. Power Output - บ่งบอกปัญหาการผลิตไฟฟ้า

กรุณาระบุ:
1. Primary Cause (สาเหตุหลัก)
2. Secondary Causes (สาเหตุรอง)
3. Propagation Chain (ลำดับการเกิดปัญหา)
4. Confidence Score (คะแนนความมั่นใจ 0-100%)
"""
        
        payload = {
            "model": "claude-sonnet-4.5",
            "messages": [{"role": "user", "content": context}],
            "max_tokens": 2000,
            "temperature": 0.2
        }
        
        response = requests.post(
            f"{self.client.base_url}/chat/completions",
            headers=self.client.headers,
            json=payload,
            timeout=30
        )
        
        return response.json()["choices"][0]["message"]["content"]
    
    def _format_sensor_list(self, sensors: List[Dict]) -> str:
        formatted = []
        for sensor in sensors:
            formatted.append(
                f"- {sensor['name']}: {sensor['value']} "
                f"(เกณฑ์ปกติ: {sensor['normal_range']}, "
                f"ส่วนเบี่ยงเบน: {sensor['deviation']:.1f}%)"
            )
        return "\n".join(formatted)


ตัวอย่างการใช้งาน

attribution_engine = SensorAttributionEngine(api_client)


ข้อมูลเซ็นเซอร์ตัวอย่าง

anomaly_sensors = [
    {"name": "VIB_Z_GEARBOX", "value": 6.8, "normal_range": "0-3.5", "deviation": 94.3},
    {"name": "TEMP_GEARBOX", "value": 95.2, "normal_range": "60-80", "deviation": 38.0},
    {"name": "RPM_DEVIATION", "value": 12.3, "normal_range": "14-15", "deviation": -11.3},
]

report = attribution_engine.generate_attribution_report(
    turbine_id="WT-2024-CN-0042",
    anomaly_data={"anomalous_sensors": anomaly_sensors},
    historical_data=pd.DataFrame()  # ข้อมูลจริงจาก database
)

print("Attribution Report:")
print(report)

故障复盘: การใช้ Claude สำหรับ Fault Review อัตโนมัติ

หลังจากเกิดความผิดปกติ การ复盘 (Review) คือกระบวนการวิเคราะห์ย้อนหลังเพื่อหาบทเรียน โมเดล Claude Sonnet 4.5 เหมาะมากสำหรับงานนี้เพราะสามารถเข้าใจบริบททางเทคนิคของ Wind Turbine ได้ดี

from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime

@dataclass
class FaultEvent:
    event_id: str
    turbine_id: str
    timestamp: datetime
    fault_code: str
    description: str
    sensor_readings: Dict
    actions_taken: List[str]
    resolution_time_minutes: int
    cost_impact: float
    weather_conditions: Dict

class FaultReviewAgent:
    """
    AI Agent สำหรับทำ Fault Review (故障复盘) อัตโนมัติ
    ใช้ Claude Sonnet 4.5 สำหรับ Root Cause Analysis และ Lessons Learned
    """
    
    def __init__(self, api_client: HolySheepWindOM):
        self.client = api_client
    
    def conduct_fault_review(self, fault_event: FaultEvent) -> Dict:
        """
        ดำเนินการ Fault Review อย่างครบวงจร
        ส่งข้อมูลให้ Claude วิเคราะห์และสรุปบทเรียน
        """
        
        prompt = f"""คุณคือ Principal Wind Turbine Engineer ที่มีประสบการณ์ 15 ปี
ทำการ Fault Review (故障复盘) สำหรับเหตุการณ์ต่อไปนี้:

=== ข้อมูลเหตุการณ์ ===
Event ID: {fault_event.event_id}
Turbine: {fault_event.turbine_id}
เวลา: {fault_event.timestamp}
Fault Code: {fault_event.fault_code}
รายละเอียด: {fault_event.description}

=== ข้อมูลเซ็นเซอร์ในขณะเกิดเหตุ ===
{self._format_sensor_data(fault_event.sensor_readings)}

=== การดำเนินการที่ทำ ===
{chr(10).join(f"- {action}" for action in fault_event.actions_taken)}

=== ผลกระทบ ===
- เวลาแก้ไข: {fault_event.resolution_time_minutes} นาที
- ต้นทุน: ${fault_event.cost_impact:,.2f}

=== สภาพอากาศ ===
- ความเร็วลม: {fault_event.weather_conditions.get('wind_speed')} m/s
- อุณหภูมิ: {fault_event.weather_conditions.get('temperature')} °C

กรุณาให้รายงานในรูปแบบต่อไปนี้:


1. Root Cause Analysis (สาเหตุที่แท้จริง)

[วิเคราะห์สาเหตุหลักและสาเหตุรอง]


2. Timeline Analysis (การวิเคราะห์ตามลำดับเวลา)

[อธิบายว่าเหตุการณ์เกิดขึ้นอย่างไรตามลำดับ]


3. Response Effectiveness (ประสิทธิภาพการตอบสนอง)

[ประเมินว่าการดำเนินการที่ทำมีประสิทธิภาพเพียงใด]


4. Lessons Learned (บทเรียนที่ได้รับ)


What Went Well:

- [รายการสิ่งที่ทำได้ดี]


What Could Be Improved:

- [รายการสิ่งที่ควรปรับปรุง]


5. Preventive Actions (มาตรการป้องกัน)

[รายการมาตรการที่ควรดำเนินการเพื่อป้องกันการเกิดซ้ำ]


6. Predictive Maintenance Recommendations

[คำแนะนำสำหรับการบำรุงรักษาเชิงคาดการณ์]
"""
        
        # เรียกใช้ Claude Sonnet 4.5 สำหรับ Fault Review
        payload = {
            "model": "claude-sonnet-4.5",
            "messages": [
                {"role": "system", "content": "คุณคือผู้เชี่ยวชาญด้านกังหันลมที่ทำ Fault Review อย่างมืออาชีพ"},
                {"role": "user", "content": prompt}
            ],
            "max_tokens": 3000,
            "temperature": 0.3
        }
        
        response = requests.post(
            f"{self.client.base_url}/chat/completions",
            headers=self.client.headers,
            json=payload,
            timeout=45
        )
        
        review_result = response.json()["choices"][0]["message"]["content"]
        
        return {
            "event_id": fault_event.event_id,
            "review_report": review_result,
            "model_used": "claude-sonnet-4.5",
            "timestamp": datetime.now().isoformat()
        }
    
    def batch_review(self, fault_events: List[FaultEvent]) -> List[Dict]:
        """ทำ Fault Review หลายเหตุการณ์พร้อมกัน"""
        results = []
        for event in fault_events:
            try:
                result = self.conduct_fault_review(event)
                results.append(result)
            except Exception as e:
                results.append({
                    "event_id": event.event_id,
                    "error": str(e),
                    "status": "failed"
                })
        return results
    
    def _format_sensor_data(self, sensor_data: Dict) -> str:
        lines = []
        for key, value in sensor_data.items():
            lines.append(f"- {key}: {value}")
        return "\n".join(lines)


ตัวอย่างการใช้งาน

review_agent = FaultReviewAgent(api_client)

fault = FaultEvent(
    event_id="FE-2024-0042",
    turbine_id="WT-2024-CN-0042",
    timestamp=datetime(2024, 3, 15, 14, 30),
    fault_code="Gbx-OverTemp-02",
    description="อุณหภูมิเกียร์บ็อกซ์สูงเกินเกณฑ์ 90°C",
    sensor_readings={
        "TEMP_GEARBOX": 96.5,
        "VIB_GEARBOX_Z": 5.8,
        "OIL_PRESSURE": 2.1,
        "RPM": 13.8
    },
    actions_taken=[
        "Reduced turbine output to 70%",
        "Activated cooling system",
        "Scheduled maintenance for next day",
        "Replaced oil filter"
    ],
    resolution_time_minutes=180,
    cost

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง