บทความนี้จะพาคุณไปรู้จักกับระบบ AI สำหรับงานบริหารจัดการระบบประปาอัจฉริยะ ที่ช่วยตรวจจับจุดรั่วไหลของท่อประปาแบบเรียลไทม์ พร้อมสร้างเอกสารประวัติการซ่อมแซมที่สอดคล้องกับมาตรฐานกำกับดูแล
กรณีศึกษา: บริษัท การประปาส่วนภูมิภาคแห่งหนึ่งในภาคตะวันออกเฉียงเหนือ
บริบทธุรกิจ
ทีมพัฒนา AI ของบริษัทการประปาแห่งหนึ่งในจังหวัดขอนแก่น ดูแลระบบท่อประปาความยาวกว่า 2,800 กิโลเมตร กระจายอยู่ในเขตพื้นที่ 14 อำเภอ ปริมาณน้ำที่ผลิตได้วันละ 185,000 ลูกบาศก์เมตร แต่อัตราการสูญเสียน้ำ (Non-Revenue Water) อยู่ที่ 28.3% ซึ่งสูงกว่าค่าเฉลี่ยของการประปาท้องถิ่นอย่างมีนัยสำคัญ
จุดเจ็บปวดของระบบเดิม
ก่อนหน้านี้ ทีมไอทีของบริษัทใช้วิธีการตรวจจับการรั่วไหลแบบดั้งเดิม โดยพึ่งพาการลงพื้นที่สำรวจของช่างเทคนิคเป็นหลัก วิธีนี้มีข้อจำกัดหลายประการ:
- ความล่าช้าในการตรวจจับ: ระยะเวลาเฉลี่ยจากจุดรั่วเกิดขึ้นจนถูกรายงานอยู่ที่ 72 ชั่วโมง ทำให้สูญเสียน้ำปริมาณมาก
- ค่าใช้จ่ายสูง: ค่าจ้างช่างลงพื้นที่สำรวจเดือนละ 280,000 บาท ยังไม่รวมค่าเดินทางและอุปกรณ์
- เอกสารไม่เป็นมาตรฐาน: การบันทึกข้อมูลการซ่อมแซมทำผ่านสมุดสมุดลูกค้าและภาพถ่าย ไม่มีระบบรวมศูนย์
- ไม่ตรงตามกฎหมาย: กรมส่งเสริมการปกครองท้องถิ่นเรียกร้องให้มีบันทึกข้อมูลการบำรุงรักษาระบบประปาตามมาตรฐาน ISO 9001 ซึ่งระบบเดิมไม่รองรับ
การเลือก HolySheep AI
หลังจากประเมินทางเลือกหลายรายการ ทีมงานตัดสินใจเลือก HolySheep AI เนื่องจาก:
- สนับสนุน DeepSeek V4 ซึ่งมีความสามารถในการวิเคราะห์ข้อมูลอนุกรมเวลา (Time Series) ได้ดีเยี่ยม ราคาเพียง $0.42/MTok
- สามารถใช้ Claude สำหรับการสร้างเอกสารประวัติการซ่อมแซมที่สอดคล้องกับมาตรฐานกำกับดูแล โดยค่าใช้จ่ายเพียง $15/MTok
- มี Webhook สำหรับรับข้อมูลจากเซ็นเซอร์ IoT ของระบบประปาโดยตรง
- รองรับการชำระเงินผ่าน WeChat Pay / Alipay ซึ่งสะดวกสำหรับการทำธุรกรรมระหว่างประเทศ
- มี Latency ต่ำกว่า 50ms ทำให้การแจ้งเตือนแบบเรียลไทม์เป็นไปได้
ขั้นตอนการย้ายระบบ
1. การเตรียมข้อมูลเซ็นเซอร์
ทีมพัฒนาได้สร้าง Data Pipeline สำหรับส่งข้อมูลจากเซ็นเซอร์วัดอัตราการไหล (Flow Meter) และเซ็นเซอร์วัดความดัน (Pressure Sensor) ไปยัง Webhook ของ HolySheep
2. การปรับ Base URL
ปรับเปลี่ยน Endpoint ของโค้ดเดิมจากระบบ AI เดิมให้ใช้ https://api.holysheep.ai/v1
3. Canary Deployment
ทีมเลือกใช้กลยุทธ์ Canary Deploy โดยเปิดระบบใหม่ให้กับพื้นที่ทดลอง 3 อำเภอก่อน (จากทั้งหมด 14 อำเภอ) เป็นระยะเวลา 2 สัปดาห์
ผลลัพธ์หลัง 30 วัน
| ตัวชี้วัด | ก่อนใช้งาน | หลังใช้งาน | การปรับปรุง |
|---|---|---|---|
| ระยะเวลาตรวจจับจุดรั่วเฉลี่ย | 72 ชั่วโมง | 4.2 ชั่วโมง | ลดลง 94.2% |
| อัตราสูญเสียน้ำ (NRW) | 28.3% | 17.8% | ลดลง 10.5% |
| ค่าสูญเสียน้ำต่อเดือน | 1.85 ล้านบาท | 680,000 บาท | ประหยัด 1.17 ล้านบาท |
| ค่าใช้จ่ายด้านบุคลากรลงพื้นที่ | 280,000 บาท/เดือน | 45,000 บาท/เดือน | ลดลง 84% |
| เวลาตอบสนอง API | 420ms | 180ms | เร็วขึ้น 57% |
| ค่าใช้จ่าย API รายเดือน | $4,200 | $680 | ประหยัด 84% |
| ความสอดคล้องกับ ISO 9001 | 62% | 97% | ผ่านการตรวจรับรอง |
วิธีการติดตั้ง Leak Detection Agent
ส่วนที่ 1: การวิเคราะห์ข้อมูลอนุกรมเวลาด้วย DeepSeek V4
import requests
import json
from datetime import datetime
class WaterLeakDetector:
"""
ระบบตรวจจับการรั่วไหลของท่อประปา
ใช้ DeepSeek V4 สำหรับวิเคราะห์ความผิดปกติในข้อมูลอนุกรมเวลา
"""
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_flow_anomaly(
self,
sensor_id: str,
flow_data: list[dict],
pressure_data: list[dict]
) -> dict:
"""
วิเคราะห์ความผิดปกติของอัตราการไหลและความดัน
ราคา: $0.42/MTok (DeepSeek V3.2)
"""
# รวมข้อมูลเซ็นเซอร์เป็นรูปแบบที่โมเดลเข้าใจได้
time_series_prompt = self._build_time_series_prompt(
sensor_id, flow_data, pressure_data
)
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "deepseek-v3.2",
"messages": [
{
"role": "system",
"content": """คุณคือผู้เชี่ยวชาญด้านการวิเคราะห์ข้อมูลระบบประปา
คุณมีความสามารถในการตรวจจับความผิดปกติของอัตราการไหลและความดันที่บ่งชี้ถึงการรั่วไหลของท่อ
ให้วิเคราะห์ข้อมูลและระบุ:
1. ความน่าจะเป็นที่มีการรั่วไหล (0-100%)
2. ตำแหน่งที่สันนิษฐานว่ารั่ว
3. ระดับความรุนแรง (ต่ำ/กลาง/สูง/วิกฤต)"""
},
{
"role": "user",
"content": time_series_prompt
}
],
"temperature": 0.3, # ค่าต่ำเพื่อความแม่นยำ
"max_tokens": 500
}
)
result = response.json()
return {
"sensor_id": sensor_id,
"analysis": result["choices"][0]["message"]["content"],
"usage": result.get("usage", {}),
"latency_ms": response.elapsed.total_seconds() * 1000
}
def _build_time_series_prompt(
self,
sensor_id: str,
flow_data: list,
pressure_data: list
) -> str:
"""สร้าง Prompt สำหรับการวิเคราะห์ข้อมูลอนุกรมเวลา"""
flow_summary = self._summarize_data_points(flow_data, "flow_rate", "m³/h")
pressure_summary = self._summarize_data_points(pressure_data, "pressure", "bar")
return f"""ข้อมูลเซ็นเซอร์ ID: {sensor_id}
วันที่วิเคราะห์: {datetime.now().isoformat()}
ข้อมูลอัตราการไหล (48 ชั่วโมงล่าสุด)
{flow_summary}
ข้อมูลความดัน (48 ชั่วโมงล่าสุด)
{pressure_summary}
โปรดวิเคราะห์ว่ามีสัญญาณการรั่วไหลหรือไม่"""
def _summarize_data_points(self, data: list, field: str, unit: str) -> str:
"""สร้างสรุปข้อมูล 48 จุด"""
lines = []
for point in data[-48:]: # 48 ชั่วโมงล่าสุด
timestamp = point.get("timestamp", "")
value = point.get(field, 0)
lines.append(f"- {timestamp}: {value:.2f} {unit}")
return "\n".join(lines)
ตัวอย่างการใช้งาน
detector = WaterLeakDetector(api_key="YOUR_HOLYSHEEP_API_KEY")
ข้อมูลจากเซ็นเซอร์ (ดึงจากระบบ SCADA ของโรงประปา)
sample_flow_data = [
{"timestamp": "2026-05-29T00:00:00Z", "flow_rate": 127.5},
{"timestamp": "2026-05-29T01:00:00Z", "flow_rate": 124.2},
{"timestamp": "2026-05-29T02:00:00Z", "flow_rate": 121.8},
# ... ข้อมูล 48 ชั่วโมง
]
sample_pressure_data = [
{"timestamp": "2026-05-29T00:00:00Z", "pressure": 3.8},
{"timestamp": "2026-05-29T01:00:00Z", "pressure": 3.9},
{"timestamp": "2026-05-29T02:00:00Z", "pressure": 2.1}, # ความดันตกผิดปกติ
# ... ข้อมูล 48 ชั่วโมง
]
result = detector.analyze_flow_anomaly(
sensor_id="FLOW-METER-001",
flow_data=sample_flow_data,
pressure_data=sample_pressure_data
)
print(f"ความน่าจะเป็นการรั่ว: {result['analysis']}")
print(f"Latency: {result['latency_ms']:.0f}ms")
print(f"ค่าใช้จ่าย: ${result['usage'].get('total_tokens', 0) * 0.00042:.4f}")ส่วนที่ 2: การสร้างเอกสารประวัติการซ่อมด้วย Claude
import requests
import json
from datetime import datetime
from typing import Optional
class WorkOrderGenerator:
"""
ระบบสร้างเอกสารประวัติการซ่อมแซมอัตโนมัติ
ใช้ Claude สำหรับการสร้างเอกสารที่สอดคล้องกับมาตรฐาน ISO 9001
ราคา: $15/MTok (Claude Sonnet 4.5)
"""
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 generate_compliance_work_order(
self,
leak_detection_result: dict,
technician_info: dict,
repair_details: Optional[dict] = None
) -> dict:
"""
สร้างเอกสารประวัติการซ่อมแซมที่สอดคล้องกับมาตรฐานกำกับดูแล
"""
system_prompt = """คุณคือผู้เชี่ยวชาญด้านการจัดการคุณภาพและการปฏิบัติตามกฎระเบียบ
ของระบบประปา คุณมีความเชี่ยวชาญในมาตรฐาน ISO 9001 และ 14001
ให้คุณสร้างเอกสารประวัติการซ่อมแซม (Work Order) ที่มีโครงสร้างดังนี้:
1. ข้อมูลทั่วไป (เลขที่ใบสั่งงาน, วันที่, ผู้รับผิดชอบ)
2. สรุปการตรวจพบปัญหา (จากผลการวิเคราะห์ AI)
3. รายละเอียดการดำเนินการซ่อมแซม
4. การประเมินผลและข้อเสนอแนะ
5. ลายเซ็นผู้รับผิดชอบและผู้ตรวจสอบ
เอกสารต้องเป็นภาษาไทย ใช้รูปแบบที่เป็นทางการ และสามารถพิมพ์เป็น PDF ได้"""
user_prompt = f"""## ข้อมูลการตรวจพบการรั่วไหล
{json.dumps(leak_detection_result, ensure_ascii=False, indent=2)}
ข้อมูลช่างเทคนิค
- ชื่อ: {technician_info.get('name', 'นายสมชาย ใจดี')}
- แผนก: {technician_info.get('department', 'แผนกบำรุงรักษา')}
- เลขที่ใบอนุญาต: {technician_info.get('license_no', 'TH-PLUMBER-2026-0042')}
รายละเอียดการซ่อม (ถ้ามี)
{json.dumps(repair_details, ensure_ascii=False, indent=2) if repair_details else 'ยังไม่ได้ดำเนินการซ่อม'}
โปรดสร้างเอกสารประวัติการซ่อมแซมตามรูปแบบที่กำหนด"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={
"model": "claude-sonnet-4.5",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.2,
"max_tokens": 2000
}
)
result = response.json()
return {
"document_id": f"WO-{datetime.now().strftime('%Y%m%d')}-{leak_detection_result.get('sensor_id', 'UNKNOWN')}",
"content": result["choices"][0]["message"]["content"],
"compliance_score": self._calculate_compliance_score(result["choices"][0]["message"]["content"]),
"usage": result.get("usage", {})
}
def _calculate_compliance_score(self, document: str) -> float:
"""
ประเมินคะแนนความสอดคล้องกับมาตรฐาน ISO
"""
required_sections = [
"เลขที่ใบสั่งงาน",
"วันที่",
"รายละเอียด",
"ลายเซ็น"
]
found_sections = sum(1 for section in required_sections if section in document)
return (found_sections / len(required_sections)) * 100
def batch_generate_reports(self, leak_results: list) -> list:
"""
สร้างเอกสารหลายฉบับพร้อมกัน
"""
work_orders = []
for result in leak_results:
technician = {
"name": result.get("assigned_technician", "นายทดสอบ ระบบ"),
"department": "แผนกบำรุงรักษาท่อประปา",
"license_no": f"TH-PLUMBER-{datetime.now().year}-{hash(result['sensor_id']) % 10000:04d}"
}
work_order = self.generate_compliance_work_order(
leak_detection_result=result,
technician_info=technician
)
work_orders.append(work_order)
return work_orders
ตัวอย่างการใช้งาน
work_order_gen = WorkOrderGenerator(api_key="YOUR_HOLYSHEEP_API_KEY")
ผลการตรวจจับจาก DeepSeek
leak_detection = {
"sensor_id": "FLOW-METER-001",
"analysis": "พบความผิดปกติของความดันที่ลดลง 45% ตั้งแต่ 02:00 น. สันนิษฐานว่ามีการรั่วที่ท่อประธานขนาด 300mm",
"probability": 94,
"severity": "สูง",
"location": "ถนนมิตรภาพ กม. 12+500"
}
technician = {
"name": "นายวิชัย ชำนาญท่อประปา",
"department": "แผนกบำรุงรักษาฝ่ายเหนือ",
"license_no": "TH-PLUMBER-2026-0089"
}
work_order = work_order_gen.generate_compliance_work_order(
leak_detection_result=leak_detection,
technician_info=technician
)
print(f"เลขที่เอกสาร: {work_order['document_id']}")
print(f"คะแนนความสอดคล้อง: {work_order['compliance_score']:.0f}%")
print(f"ค่าใช้จ่าย: ${work_order['usage'].get('total_tokens', 0) * 0.000015:.4f}")ส่วนที่ 3: Webhook สำหรับรับข้อมูลจากเซ็นเซอร์ IoT
from flask import Flask, request, jsonify
import hashlib
import hmac
import logging
from datetime import datetime
import threading
app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
ตั้งค่า Webhook Secret สำหรับตรวจสอบความถูกต้องของข้อมูล
WEBHOOK_SECRET = "YOUR_WEBHOOK_SECRET"
BATCH_PROCESSING = True
batch_data = []
batch_lock = threading.Lock()
class IoTWebhookHandler:
"""
ระบบรับข้อมูลจากเซ็นเซอร์ IoT ผ่าน Webhook
รองรับการรับข้อมูลแบบ Real-time และ Batch
"""
def __init__(self, holysheep_api_key: str):
self.api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
def verify_signature(self, payload: bytes, signature: str) -> bool:
"""ตรวจสอบความถูกต้องของ Webhook Signature"""
expected_signature = hmac.new(
WEBHOOK_SECRET.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected_signature}", signature)
def process_sensor_data(self, sensor_data: dict) -> dict:
"""
ประมวลผลข้อมูลจากเซ็นเซอร์และเรียกใช้ Leak Detection API
"""
import requests
# เรียกใช้ Leak Detection
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [
{
"role": "system",
"content": "วิเคราะห์ข้อมูลเซ็นเซอร์ระบบประปา ระบุความผิดปกติ"
},
{
"role": "user",
"content": f"ข้อมูลเซ็นเซอร์: {sensor_data}"
}
]
}
)
return {
"status": "processed",
"sensor_id": sensor_data.get("sensor_id"),
"result": response.json(),
"timestamp": datetime.now().isoformat()
}
webhook_handler = IoTWebhookHandler(api_key="YOUR_HOLYSHEEP_API_KEY")
@app.route('/webhook/sensor', methods=['POST'])
def receive_sensor_data():
"""
Endpoint สำหรับรับข้อมูลจากเซ็นเซอร์ IoT
Content-Type: application/json
"""
# ตรวจสอบ Webhook Signature
signature = request.headers.get('X-Webhook-Signature', '')
if not webhook_handler.verify_signature(request.data, signature):
logger.warning("Webhook signature