OpenClaw：从“瞎养”到“虾养你”——对虾养殖全流程自动化

原创

用户12339161

发布于 2026-10-22 14:33:49

160

对虾养殖最大的浪费不是饲料，而是“瞎”：凭经验看水色、靠手摸估投喂、增氧机常开常闭拍脑袋。一场弧菌暴发，一夜之间清塘。OpenClaw 是一套开源的养殖控制平台，把传感、边缘判断、闭环控制、风险预警串成一条流水线，让每一度电、每一克料都由数据决定。本文拆解它的四层实现与关键代码。

一、四层架构

传感层：溶氧、pH、水温、氨氮、亚硝酸盐、盐度、ORP 多参数探头，Modbus-RTU 采集后转 MQTT 上云；
边缘层：塘口网关做数据清洗、断线补传、阈值分级，断网也能本地控制；
决策层：投喂量、增氧时长、换水策略的闭环计算；
风险层：病害预警与用药建议。

边缘层承担“断网自治”，是整套系统能在真实塘口活下来的前提。

二、传感接入：MQTT + 断线补传

探头每 30 秒上报一次，网关订阅后落库。关键在“补传”：网络恢复时把缓冲区数据按时间戳写回，避免曲线断裂误判。

import paho.mqtt.client as mqtt, json, time, sqlite3

BUF = sqlite3.connect("edge.db")
BUF.execute("CREATE TABLE IF NOT EXISTS buf(ts INT, pool TEXT, k TEXT, v REAL)")

def on_message(c, u, msg):
    d = json.loads(msg.payload)
    row = {
        "ts": int(time.time()), "pool": d["pool"],
        "do": d["do"], "ph": d["ph"], "temp": d["temp"],
        "nh3": d["nh3"], "no2": d["no2"], "sal": d["sal"],
    }
    try:
        influx.write("water", row)          # 在线直写
    except Exception:
        for k, v in row.items():
            if isinstance(v, (int, float)):
                BUF.execute("INSERT INTO buf VALUES(?,?,?,?)",
                            (row["ts"], row["pool"], k, v))
        BUF.commit()

def flush_buffer():                          # 定时补传
    for ts, pool, k, v in BUF.execute("SELECT * FROM buf").fetchall():
        influx.write("water", {"ts": ts, "pool": pool, k: v})
    BUF.execute("DELETE FROM buf"); BUF.commit()

client = mqtt.Client()
client.on_message = on_message
client.connect("broker.local", 1883)
client.subscribe("pond/+/sensor")
client.loop_start()

三、水质分级：带滞回的阈值判断

直接比阈值会在临界点来回抖动，导致告警刷屏、增氧机频繁启停。正确做法是双阈值 + 滞回：进入危险态需跌破下限，退出需回升到安全线以上。

# 每项: (安全下限, 危险下限, 安全上限, 危险上限)
TH = {
  "do":  (5.0, 3.0, None, None),
  "ph":  (7.0, 6.5, 8.5, 9.0),
  "nh3": (None, None, 0.2, 0.5),   # 越低越好
  "no2": (None, None, 0.1, 0.3),
}

def grade(state, k, v):
    lo_s, lo_d, hi_s, hi_d = TH[k]
    if lo_d is not None and v < lo_d: return "danger"
    if lo_s is not None and v < lo_s: return "warn"
    if hi_d is not None and v > hi_d: return "danger"
    if hi_s is not None and v > hi_s: return "warn"
    return "ok"

def evaluate(pool, samples):
    for k, v in samples.items():
        nxt = grade(None, k, v)
        if nxt != state[pool][k]:        # 仅在状态跳变时触发动作
            state[pool][k] = nxt
            act(pool, k, nxt, v)

四、闭环控制：增氧机与投喂

增氧机按溶氧做比例控制，避免“常开费电、关早缺氧”：

def aerator(pool, do):
    duty = clamp((5.0 - do) / 2.0, 0, 1)   # DO=5 关，DO=3 全开
    pwm.set(pool, "aerator", duty)

投喂由水温、溶氧、摄食史共同决定，而不是固定钟点。日投喂率随水温先升后降，溶氧不足时减量甚至停喂——虾在低氧下吃料等于自杀。

def feed_ratio(temp, do):
    base = {-1: 0}
    if temp < 18: return 0.0
    r = 0.03 if temp < 24 else 0.045 if temp < 30 else 0.025
    return r * clamp(do / 5.0, 0, 1)       # 低氧打折

def feed_once(pool, biomass_kg, temp, do):
    kg = biomass_kg * feed_ratio(temp, do)
    feeder.dispense(pool, round(kg, 2))
    db.log(pool, kg, temp, do)

五、病害预警：把水质变“分”

弧菌、白斑病暴发前，水质曲线通常先异常。用一组加权规则把多参数合成为 0–100 的风险分，超过 70 触发“减料 + 增氧 + 抽检”预案：

def risk_score(s):
    r = 0
    r += 30 if s["do"] < 3.5 else 0
    r += 25 if s["nh3"] > 0.3 else 0
    r += 25 if s["no2"] > 0.2 else 0
    r += 10 if not (7.5 <= s["ph"] <= 8.6) else 0
    r += 10 if s["temp"] > 32 else 0
    return min(r, 100)

if risk_score(samples) >= 70:
    reduce_feed(pool, 0.5); aerator(pool, 3.0)
    notify("病害高风险", pool, samples)

六、全流程串起来

一条完整的养殖日流程：探头 30 秒一次上报 → 网关分级判定 → 溶氧跌破 3 触发增氧全开并告警 → 投喂前按当前水温溶氧算当日料量 → 夜间每 2 小时算一次风险分 → 连续两天高分自动转“减料+抽检”预案。养殖户看到的不是几百个数字，而是三件事：今天喂多少、机器怎么转、有没有风险。

“让虾养你”的本质不是无人值守，而是把人的经验固化成可回溯的规则与曲线。数据跑通了，虾的存活率、料肉比、电费都变成可优化的变量——到那时，虾确实在替你干活。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

玩转OpenClaw云端创意实践