智慧生活科技專業社群: AIoT智慧養殖實戰：整合Django、Spark與AI模型進行水質異常預警

智慧生活科技專業社群｜物聯網 × 大數據 × AIoT 系列（七）

前一篇我們讓 Spark Structured Streaming 的水質統計顯示在 Django Dashboard。本篇要再往前走一步：利用機器學習找出不尋常的水質變化，並將結果顯示在網頁上。

情境以智慧養殖池為例，數據均為教學模擬資料，並非水井村實際量測或經驗證的養殖安全門檻。

一、從監控走向預警：為什麼需要AI？

傳統規則可以判斷「溶氧低於某個門檻」，但對於多個指標同時產生細微變化的情況，機器學習可提供額外線索。例如水溫上升、溶氧下降且pH變化異常，單一指標可能尚未超出設定範圍，整體組合卻與過去不同。

方法	判斷方式	注意事項
門檻規則	依專家設定範圍比較	不同物種、養殖階段與設備需重新確認
Isolation Forest	找出相對於訓練資料少見的組合	異常不等於危險，模型可能誤報或漏報
人工確認	查核感測器、現場狀況及歷史趨勢	不可省略

二、系統架構與分工

ESP32／模擬感測器 → MQTT → Python Bridge → Spark Structured Streaming → SQLite（視窗統計） → AI Worker → SQLite（AI警示） → Django API → Dashboard

Spark負責大量資料的時間視窗統計；Scikit-learn負責模型訓練與推論；Django負責展示。此教學版本讓AI Worker獨立執行，避免每次網頁請求都重新訓練模型。

三、準備環境與資料表

假設上一篇已建立Django專案 aquamon 與應用程式 monitor。以下使用同一個SQLite資料庫。為了讓範例能獨立理解，先定義Spark輸出的資料格式：

資料表：monitor_windowstat	意義
id	整數主鍵
pond_id	養殖池代號
window_start	視窗開始時間（UTC）
temperature_avg	平均水溫
ph_avg	平均pH
oxygen_avg	平均溶氧

若前一篇的模型名稱或欄位不同，請先調整為一致；下列程式不會自動建立既有統計資料表。

python -m pip install django pandas numpy scikit-learn joblib
# Spark / MQTT 元件沿用上一篇環境；本篇 AI Worker 不必啟動 Spark。

四、Django新增AI警示模型

在 monitor/models.py 保留既有模型，並新增以下類別。source_stat_id 用來避免同一筆視窗統計被重複判斷。

from django.db import models

class AnomalyAlert(models.Model):
    source_stat_id = models.IntegerField(unique=True)
    pond_id = models.CharField(max_length=30)
    window_start = models.DateTimeField()
    anomaly_score = models.FloatField()
    is_anomaly = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["-window_start"]

python manage.py makemigrations monitor
python manage.py migrate

五、建立模擬訓練資料與Isolation Forest模型

建立 train_model.py。先用模擬的正常水質資料訓練模型；實際部署時應改用經人工確認、涵蓋不同季節與操作情境的歷史資料。模型檔案僅應載入自己產生且可信任的檔案。

# train_model.py
import numpy as np
import pandas as pd
import joblib
from sklearn.ensemble import IsolationForest

rng = np.random.default_rng(42)
n = 1000
training = pd.DataFrame({
    "temperature_avg": rng.normal(28, 1.2, n),
    "ph_avg": rng.normal(7.8, 0.25, n),
    "oxygen_avg": rng.normal(5.8, 0.65, n),
})
features = ["temperature_avg", "ph_avg", "oxygen_avg"]
model = IsolationForest(
    n_estimators=150, contamination=0.03, random_state=42
)
model.fit(training[features])
joblib.dump({"model": model, "features": features}, "water_model.joblib")
print("模型訓練完成：water_model.joblib")

contamination=0.03 是對訓練資料異常比例的假設，不是預測準確率，也不是水質危險機率。此模型使用三個視窗平均值，尚未納入季節、投餌、日夜或感測器漂移等因素。

六、AI Worker：讀取Spark統計結果並產生警示

建立 ai_worker.py，放在與 manage.py 相同的資料夾。這個程式定期讀取尚未處理的視窗統計資料，以模型計算異常分數，再將結果寫入Django資料表。

# ai_worker.py
import os
import time
import joblib
import pandas as pd

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "aquamon.settings")
import django
django.setup()

from monitor.models import WindowStat, AnomalyAlert

bundle = joblib.load("water_model.joblib")  # 僅載入可信任的模型檔
model = bundle["model"]
features = bundle["features"]

while True:
    rows = WindowStat.objects.order_by("id")[:1000]
    processed = set(AnomalyAlert.objects.values_list(
        "source_stat_id", flat=True
    ))
    pending = [r for r in rows if r.id not in processed]

    for r in pending:
        sample = pd.DataFrame([{
            "temperature_avg": r.temperature_avg,
            "ph_avg": r.ph_avg,
            "oxygen_avg": r.oxygen_avg,
        }])[features]
        if sample.isna().any().any():
            continue
        score = float(model.decision_function(sample)[0])
        abnormal = bool(model.predict(sample)[0] == -1)
        AnomalyAlert.objects.get_or_create(
            source_stat_id=r.id,
            defaults={
                "pond_id": r.pond_id,
                "window_start": r.window_start,
                "anomaly_score": score,
                "is_anomaly": abnormal,
            },
        )
        print(r.pond_id, r.window_start, score, abnormal)
    time.sleep(5)

教學版限制：上述程式每輪只查看最早的1000筆統計資料，適合少量資料驗證；長期運作應改用資料庫查詢排除已處理資料、游標或任務佇列，避免新資料被舊資料擋住。SQLite也不適合大量並行寫入，正式系統建議使用PostgreSQL並加入重試、日誌與權限管理。

七、Django API：查詢最新異常警示

在 monitor/views.py 新增API：

from django.http import JsonResponse
from monitor.models import AnomalyAlert

def alerts_api(request):
    rows = AnomalyAlert.objects.order_by("-window_start")[:30]
    data = [{
        "pond_id": r.pond_id,
        "window_start": r.window_start.isoformat(),
        "anomaly_score": round(r.anomaly_score, 4),
        "is_anomaly": r.is_anomaly,
    } for r in rows]
    return JsonResponse({"alerts": data})

在 monitor/urls.py 加入路由，並確認專案的主URL已包含 monitor.urls：

from django.urls import path
from . import views

urlpatterns = [
    path("api/alerts/", views.alerts_api, name="alerts_api"),
]

八、Dashboard顯示AI警示

將以下區塊加入上一篇的Dashboard模板。若部署到正式網站，請為API加入登入驗證與適當授權。

<section>
  <h3>AI水質異常警示（教學示範）</h3>
  <p id="ai-status">正在讀取資料…</p>
  <ul id="ai-alerts"></ul>
</section>
<script>
async function refreshAlerts() {
  const status = document.getElementById('ai-status');
  const list = document.getElementById('ai-alerts');
  try {
    const response = await fetch('/api/alerts/');
    if (!response.ok) throw new Error('API讀取失敗');
    const result = await response.json();
    list.replaceChildren();
    const abnormal = result.alerts.filter(a => a.is_anomaly);
    status.textContent = `最近30筆分析中有 ${abnormal.length} 筆模型異常`;
    for (const item of abnormal) {
      const li = document.createElement('li');
      li.textContent = `${item.pond_id}｜${item.window_start}｜分數 ${item.anomaly_score}`;
      list.appendChild(li);
    }
  } catch (err) {
    status.textContent = '暫時無法取得AI分析結果';
  }
}
refreshAlerts();
setInterval(refreshAlerts, 10000);
</script>

九、啟動與測試順序

沿用上一篇啟動MQTT Broker、感測器模擬程式、Python Bridge及Spark統計程式，確認資料寫入 WindowStat。
執行 python train_model.py，產生模型檔。
執行 python ai_worker.py，觀察推論結果。
執行 python manage.py runserver，開啟Dashboard與 /api/alerts/。
製造一筆偏離正常分布的測試統計資料，確認API與頁面能顯示警示。

測試資料可透過Django shell插入（請依實際模型欄位調整）：

python manage.py shell
>>> from django.utils import timezone
>>> from monitor.models import WindowStat
>>> WindowStat.objects.create(
...     pond_id="POND-TEST", window_start=timezone.now(),
...     temperature_avg=38.0, ph_avg=5.0, oxygen_avg=1.0
... )

此筆數據只是測試用的明顯偏離值，不代表任何養殖物種的通用危險門檻。若模型未標記為異常，應檢查訓練資料、欄位順序與模型輸出，而非任意宣稱系統已正確預警。

十、如何評估AI模型是否可靠？

正式部署前，應蒐集人工標記的正常與異常事件，評估精確率（Precision）、召回率（Recall）、誤報率及漏報情況，並依不同養殖池、季節與感測器型號分開驗證。若沒有真實異常標籤，就不能聲稱模型達到特定預測準確率。

此外，應先確認感測器是否校正、資料是否延遲、是否存在斷線後補傳，以及異常是否由維修或清洗造成。AI預警宜作為人員巡查與判斷的輔助，不應直接觸發高風險設備操作。

十一、學生延伸挑戰

加入「單一指標門檻警示」，比較規則式警示與AI異常偵測的差異。
加入水質變化率、前一時段均值等特徵，觀察是否有助於辨識異常。
依養殖池分別訓練模型，討論共用模型與專屬模型的優缺點。
改用PostgreSQL，並為AI Worker加入增量查詢、錯誤重試與模型版本紀錄。
將異常事件連結現場查核表，建立「偵測→通知→確認→回饋」流程。

十二、回顧與下一步

本篇完成了從Spark時間視窗統計、Scikit-learn模型推論，到Django API與Dashboard展示的完整教學流程。學生可以體會：IoT負責取得資料，大數據技術負責整理資料，AI協助發現不尋常的模式，而真正的智慧決策仍需要專業知識與人工確認。

延伸閱讀｜智慧生活科技專業社群

教學提醒：本文程式為概念驗證，部署前需補上安全驗證、資料庫效能、模型監控與場域驗證。

智慧生活科技專業社群

2026年11月13日星期五

AIoT智慧養殖實戰：整合Django、Spark與AI模型進行水質異常預警

一、從監控走向預警：為什麼需要AI？

二、系統架構與分工

三、準備環境與資料表

四、Django新增AI警示模型

五、建立模擬訓練資料與Isolation Forest模型

六、AI Worker：讀取Spark統計結果並產生警示

七、Django API：查詢最新異常警示

八、Dashboard顯示AI警示

九、啟動與測試順序

十、如何評估AI模型是否可靠？

十一、學生延伸挑戰

十二、回顧與下一步

延伸閱讀｜智慧生活科技專業社群

沒有留言:

張貼留言

2026年11月13日 星期五

AIoT智慧養殖實戰：整合Django、Spark與AI模型進行水質異常預警

一、從監控走向預警：為什麼需要AI？

二、系統架構與分工

三、準備環境與資料表

四、Django新增AI警示模型

五、建立模擬訓練資料與Isolation Forest模型

六、AI Worker：讀取Spark統計結果並產生警示

七、Django API：查詢最新異常警示

八、Dashboard顯示AI警示

九、啟動與測試順序

十、如何評估AI模型是否可靠？

十一、學生延伸挑戰

十二、回顧與下一步

延伸閱讀｜智慧生活科技專業社群

沒有留言:

張貼留言

2026年11月13日星期五