การเพาะเลี้ยงสัตว์น้ำอัจฉริยะในยุคปัญญาประดิษฐ์ต้องการระบบที่คอยเฝ้าระวังค่าออกซิเจนละลายน้ำ (Dissolved Oxygen) อย่างต่อเนื่อง บทความนี้จะสอนวิธีสร้างแพลตฟอร์มเตือนภัยล่วงหน้าด้วย HolySheep AI ที่ผสาน GPT-5 สำหรับการคาดการณ์ความผิดปกติ ร่วมกับ Gemini สำหรับการวิเคราะห์ภาพคุณภาพน้ำ และการกำหนดค่า SLA การจำกัดอัตราการร้องขอ (Rate Limiting) พร้อมโค้ดตัวอย่างที่พร้อมใช้งานจริง
บทนำ: ทำไมการเพาะเลี้ยงสัตว์น้ำต้องการ AI Early Warning
ในอุตสาหกรรมการเพาะเลี้ยงสัตว์น้ำ ค่าออกซิเจนละลายน้ำต่ำกว่า 4 mg/L อาจทำให้ปลาและกุ้งเสียชีวิตภายใน 2-4 ชั่วโมง ระบบ Early Warning ที่ดีต้อง:
- คาดการณ์ล่วงหน้า 30-60 นาที ก่อนที่ค่าออกซิเจนจะลดต่ำกว่าเกณฑ์อันตราย
- วิเคราะห์ภาพพื้นที่เลี้ยง เพื่อตรวจจับการเปลี่ยนแปลงของสีน้ำ ตะกอน และความหนาแน่นของสัตว์น้ำ
- จัดการการร้องขอ API อย่างมีประสิทธิภาพเพื่อไม่ให้เกินโควต้าในช่วงวิกฤต
- ส่งการแจ้งเตือนหลายช่องทาง ทั้ง LINE, SMS และ Email
ระบบนี้เหมาะสำหรับฟาร์มกุ้ง ฟาร์มปลานิล บ่อเลี้ยงปลาคาร์ป และการเพาะเลี้ยงสัตว์น้ำในระบบปิด (RAS)
สถาปัตยกรรมระบบ HolySheep สำหรับการเตือนภัยค่าออกซิเจน
1. การติดตั้งและเริ่มต้น API Client
#!/usr/bin/env python3
"""
ระบบเตือนภัยค่าออกซิเจนละลายน้ำ - HolySheep AI Integration
การคาดการณ์ความผิดปกติด้วย GPT-5 + การวิเคราะห์ภาพด้วย Gemini
"""
import os
import time
import json
import base64
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum
import requests
import numpy as np
from PIL import Image
import io
============================================================
การกำหนดค่า HolySheep API
============================================================
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
ราคาโมเดล 2026 (USD per Million Tokens)
MODEL_PRICES = {
"gpt-5": 8.00, # GPT-4.1: $8/MTok
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
เกณฑ์ค่าออกซิเจน (mg/L)
DO_CRITICAL = 3.0 # อันตรายต่ำสุด
DO_WARNING = 4.0 # เริ่มเตือน
DO_NORMAL = 5.0 # ปกติ
@dataclass
class SensorReading:
"""ข้อมูลจากเซ็นเซอร์"""
timestamp: datetime
dissolved_oxygen: float # mg/L
temperature: float # องศาเซลเซียส
pH: float
turbidity: float # NTU
ammonia: float # mg/L
image_data: Optional[bytes] = None # ภาพพื้นที่เลี้ยง
@dataclass
class PredictionResult:
"""ผลลัพธ์การคาดการณ์"""
predicted_do: float
confidence: float
time_to_critical: int # นาทีจนถึงค่าวิกฤต
risk_level: str # "low", "medium", "high", "critical"
recommendations: List[str]
class HolySheepAIClient:
"""Client สำหรับ HolySheep AI API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
self.rate_limiter = RateLimiter(
requests_per_minute=60,
requests_per_day=10000
)
self.usage_stats = {"total_tokens": 0, "total_cost_usd": 0.0}
def _make_request(
self,
endpoint: str,
model: str,
messages: List[Dict],
max_tokens: int = 2048,
temperature: float = 0.3
) -> Dict[str, Any]:
"""ส่งคำขอไปยัง HolySheep API พร้อม Rate Limiting"""
# ตรวจสอบ Rate Limit
if not self.rate_limiter.can_make_request():
wait_time = self.rate_limiter.get_wait_time()
logging.warning(f"Rate limit reached. Waiting {wait_time:.1f}s")
time.sleep(wait_time)
self.rate_limiter.record_request()
url = f"{self.base_url}{endpoint}"
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
}
try:
response = self.session.post(url, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
# คำนวณค่าใช้จ่าย
usage = result.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = prompt_tokens + completion_tokens
cost = (total_tokens / 1_000_000) * MODEL_PRICES.get(model, 1.0)
self.usage_stats["total_tokens"] += total_tokens
self.usage_stats["total_cost_usd"] += cost
logging.info(f"API Call: {model} | Tokens: {total_tokens} | Cost: ${cost:.4f}")
return result
except requests.exceptions.RequestException as e:
logging.error(f"API Request Failed: {e}")
raise
def predict_dissolved_oxygen_anomaly(
self,
readings: List[SensorReading],
farm_context: str
) -> PredictionResult:
"""
ใช้ GPT-5 (GPT-4.1) คาดการณ์ความผิดปกติของค่าออกซิเจน
ราคา: $8/MTok
"""
# สร้าง prompt สำหรับการวิเคราะห์
readings_text = "\n".join([
f"- [{r.timestamp.strftime('%H:%M')}] DO: {r.dissolved_oxygen}mg/L, "
f"Temp: {r.temperature}°C, pH: {r.pH}, ความขุ่น: {r.turbidity}NTU, "
f"Ammonia: {r.ammonia}mg/L"
for r in readings[-10:] # 10 ค่าล่าสุด
])
system_prompt = """คุณเป็นผู้เชี่ยวชาญด้านการเพาะเลี้ยงสัตว์น้ำ วิเคราะห์ข้อมูลเซ็นเซอร์และคาดการณ์ค่าออกซิเจนละลายน้ำ
เกณฑ์ความเสี่ยง:
- DO > 5.0 mg/L: ปกติ (low)
- DO 4.0-5.0 mg/L: เริ่มกังวล (medium)
- DO 3.0-4.0 mg/L: เสี่ยง (high)
- DO < 3.0 mg/L: วิกฤต (critical)
คืนค่า JSON ที่มี:
- predicted_do: ค่าออกซิเจนที่คาดการณ์ (mg/L)
- confidence: ความมั่นใจ (0.0-1.0)
- time_to_critical: นาทีจนถึงค่าวิกฤต
- risk_level: low/medium/high/critical
- recommendations: รายการคำแนะนำ"""
user_message = f"""ข้อมูลฟาร์ม: {farm_context}
ข้อมูลเซ็นเซอร์ 10 ค่าล่าสุด:
{readings_text}
วิเคราะห์แนวโน้มและคาดการณ์ค่าออกซิเจนในอีก 30 นาทีข้างหน้า"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
]
response = self._make_request(
endpoint="/chat/completions",
model="gpt-5", # HolySheep map to GPT-4.1
messages=messages,
max_tokens=1024,
temperature=0.2
)
content = response["choices"][0]["message"]["content"]
# Parse JSON response
try:
# ลองดึง JSON จาก response
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
result_data = json.loads(content.strip())
return PredictionResult(
predicted_do=float(result_data["predicted_do"]),
confidence=float(result_data["confidence"]),
time_to_critical=int(result_data["time_to_critical"]),
risk_level=result_data["risk_level"],
recommendations=result_data.get("recommendations", [])
)
except json.JSONDecodeError:
logging.error(f"Failed to parse GPT response: {content}")
return PredictionResult(
predicted_do=readings[-1].dissolved_oxygen,
confidence=0.0,
time_to_critical=999,
risk_level="unknown",
recommendations=["ไม่สามารถวิเคราะห์ได้"]
)
def analyze_water_quality_image(
self,
image_bytes: bytes,
analysis_type: str = "full"
) -> Dict[str, Any]:
"""
ใช้ Gemini วิเคราะห์ภาพคุณภาพน้ำ
ราคา: $2.50/MTok (ประหยัดกว่า Claude 6 เท่า)
"""
# แปลงภาพเป็น base64
image_base64 = base64.b64encode(image_bytes).decode('utf-8')
system_prompt = """คุณเป็นผู้เชี่ยวชาญด้านคุณภาพน้ำสำหรับการเพาะเลี้ยงสัตว์น้ำ
วิเคราะห์ภาพและประเมิน:
1. สีของน้ำ (เขียว/น้ำตาล/ใส/ขุ่น)
2. การมีตะกอนหรือสาหร่าย
3. ความหนาแน่นของสัตว์น้ำ (ดูจากการเคลื่อนไหว/เงาที่ผิวน้ำ)
4. สัญญาณความเครียดของสัตว์น้ำ (ฮวบฮาบ/ลอยตัว/กระโดด)
คืนค่า JSON ที่มี:
- water_color: คำอธิบายสีน้ำ
- turbidity_estimate: low/medium/high
- algae_present: true/false
- algae_level: 0-100%
- fish_density: low/medium/high
- stress_signs: [list of observed signs]
- overall_health_score: 0-100
- concerns: รายการปัญหาที่พบ"""
user_message = f"วิเคราะห์ภาพคุณภาพน้ำประเภท: {analysis_type}"
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
]
response = self._make_request(
endpoint="/chat/completions",
model="gemini-2.5-flash",
messages=messages,
max_tokens=1536,
temperature=0.1
)
content = response["choices"][0]["message"]["content"]
try:
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
return json.loads(content.strip())
except json.JSONDecodeError:
logging.error(f"Failed to parse Gemini response")
return {"error": "Failed to analyze image"}
def generate_action_plan(
self,
prediction: PredictionResult,
image_analysis: Dict[str, Any]
) -> str:
"""
ใช้ DeepSeek V3.2 สร้างแผนปฏิบัติการ (ราคาถูกที่สุด: $0.42/MTok)
"""
context = f"""การคาดการณ์ค่าออกซิเจน:
- ค่าที่คาดการณ์: {prediction.predicted_do} mg/L
- ระดับความเสี่ยง: {prediction.risk_level}
- เวลาถึงวิกฤต: {prediction.time_to_critical} นาที
- คำแนะนำ: {', '.join(prediction.recommendations)}
การวิเคราะห์ภาพ:
- คะแนนสุขภาพโดยรวม: {image_analysis.get('overall_health_score', 'N/A')}
- สีน้ำ: {image_analysis.get('water_color', 'N/A')}
- สาหร่าย: {image_analysis.get('algae_present', False)} ({image_analysis.get('algae_level', 0)}%)
- สัญญาณความเครียด: {image_analysis.get('stress_signs', [])}"""
messages = [
{"role": "system", "content": "คุณเป็นที่ปรึกษาด้านการเพาะเลี้ยงสัตว์น้ำ ให้คำแนะนำที่เป็นรูปธรรมและเร่งด่วน"},
{"role": "user", "content": f"สร้างแผนปฏิบัติการ 5 ขั้นตอน:\n{context}"}
]
response = self._make_request(
endpoint="/chat/completions",
model="deepseek-v3.2",
messages=messages,
max_tokens=512,
temperature=0.5
)
return response["choices"][0]["message"]["content"]
print("✅ HolySheep AI Client initialized")
print(f"📊 Model Prices: {MODEL_PRICES}")
2. ระบบ Rate Limiting และ Retry Logic ตามมาตรฐาน SLA
"""
ระบบ Rate Limiting ตามมาตรฐาน SLA สำหรับการเตือนภัยการเพาะเลี้ยงสัตว์น้ำ
- 5xx Errors: Retry ทุก 30 วินาที สูงสุด 5 ครั้ง (Exponential Backoff)
- 429 Rate Limited: รอตาม Retry-After header
- 4xx Client Errors: ไม่ Retry และแจ้งเตือนผู้ดูแลระบบ
"""
import time
import threading
from collections import deque
from typing import Optional, Callable
from dataclasses import dataclass
import logging
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
"""การกำหนดค่า Rate Limiting ตามประเภท SLA"""
requests_per_minute: int = 60
requests_per_hour: int = 2000
requests_per_day: int = 10000
burst_allowance: int = 10
# Retry Configuration
max_retries: int = 5
base_retry_delay: float = 1.0 # วินาที
max_retry_delay: float = 60.0 # วินาที
retry_on_5xx: bool = True
retry_on_429: bool = True
class RateLimiter:
"""ระบบจำกัดอัตราการร้องขอแบบ Token Bucket"""
def __init__(self, requests_per_minute: int = 60, requests_per_day: int = 10000):
self.requests_per_minute = requests_per_minute
self.requests_per_day = requests_per_day
# ติดตาม request timestamps
self.minute_window = deque(maxlen=requests_per_minute)
self.day_window = deque(maxlen=requests_per_day)
self._lock = threading.Lock()
self.last_request_time = 0
def can_make_request(self) -> bool:
"""ตรวจสอบว่าสามารถส่ง request ได้หรือไม่"""
with self._lock:
now = time.time()
current_minute = int(now // 60)
# ลบ request เก่าออกจากหน้าต่างนาที
while self.minute_window and self.minute_window[0] < now - 60:
self.minute_window.popleft()
# ลบ request เก่าออกจากหน้าต่างวัน
while self.day_window and self.day_window[0] < now - 86400:
self.day_window.popleft()
return (
len(self.minute_window) < self.requests_per_minute and
len(self.day_window) < self.requests_per_day
)
def record_request(self):
"""บันทึกการส่ง request"""
with self._lock:
now = time.time()
self.minute_window.append(now)
self.day_window.append(now)
self.last_request_time = now
def get_wait_time(self) -> float:
"""คำนวณเวลาที่ต้องรอ (วินาที)"""
with self._lock:
now = time.time()
if not self.minute_window:
return 0.0
oldest_in_minute = self.minute_window[0]
minute_wait = max(0, 60 - (now - oldest_in_minute))
oldest_in_day = self.day_window[0] if self.day_window else now
day_wait = max(0, 86400 - (now - oldest_in_day))
return max(minute_wait, day_wait)
class RetryHandler:
"""ตัวจัดการ Retry Logic พร้อม Exponential Backoff"""
def __init__(self, config: Optional[RateLimitConfig] = None):
self.config = config or RateLimitConfig()
self.request_count = 0
self.success_count = 0
self.retry_count = 0
def calculate_backoff(self, attempt: int) -> float:
"""คำนวณเวลารอแบบ Exponential Backoff"""
delay = self.config.base_retry_delay * (2 ** attempt)
# เพิ่ม jitter ±25%
import random
jitter = delay * 0.25 * (2 * random.random() - 1)
return min(delay + jitter, self.config.max_retry_delay)
def should_retry(self, status_code: int, attempt: int) -> bool:
"""ตรวจสอบว่าควร Retry หรือไม่"""
if attempt >= self.config.max_retries:
return False
# Retry on 5xx errors
if self.config.retry_on_5xx and 500 <= status_code < 600:
return True
# Retry on 429 Rate Limited
if self.config.retry_on_429 and status_code == 429:
return True
return False
def execute_with_retry(
self,
func: Callable,
*args,
**kwargs
) -> any:
"""
Execute function with retry logic
Returns:
Tuple of (result, success, attempts, total_time)
"""
self.request_count += 1
start_time = time.time()
last_error = None
for attempt in range(self.config.max_retries + 1):
try:
result = func(*args, **kwargs)
# ตรวจสอบ HTTP status code จาก result
if hasattr(result, 'status_code'):
status_code = result.status_code
if self.should_retry(status_code, attempt):
self.retry_count += 1
wait_time = self.calculate_backoff(attempt)
# ตรวจสอบ Retry-After header
if hasattr(result, 'headers') and 'Retry-After' in result.headers:
wait_time = float(result.headers['Retry-After'])
logger.warning(
f"Attempt {attempt + 1} failed with {status_code}. "
f"Retrying in {wait_time:.1f}s"
)
time.sleep(wait_time)
continue
if 400 <= status_code < 500:
logger.error(f"Client error {status_code}: {result.text}")
raise ValueError(f"API Error: {status_code}")
elapsed = time.time() - start_time
self.success_count += 1
return {
"success": True,
"result": result,
"attempts": attempt + 1,
"elapsed_seconds": round(elapsed, 2),
"retry_count": self.retry_count
}
except Exception as e:
last_error = e
elapsed = time.time() - start_time
# ไม่ retry สำหรับ client errors
if isinstance(e, (ValueError, TypeError, KeyError)):
logger.error(f"Permanent error: {e}")
return {
"success": False,
"result": None,
"attempts": attempt + 1,
"elapsed_seconds": round(elapsed, 2),
"error": str(last_error)
}
if attempt < self.config.max_retries:
self.retry_count += 1
wait_time = self.calculate_backoff(attempt)
logger.warning(f"Attempt {attempt + 1} error: {e}. Retrying in {wait_time:.1f}s")
time.sleep(wait_time)
else:
logger.error(f"All {self.config.max_retries + 1} attempts failed")
return {
"success": False,
"result": None,
"attempts": self.config.max_retries + 1,
"elapsed_seconds": round(time.time() - start_time, 2),
"error": str(last_error)
}
def get_stats(self) -> dict:
"""สถิติการใช้งาน"""
return {
"total_requests": self.request_count,
"successful_requests": self.success_count,
"total_retries": self.retry_count,
"success_rate": round(
self.success_count / self.request_count * 100, 2
) if