水产养殖环境里，溶解氧、水温、pH值和氨氮浓度是最关键的几项指标。传感器通常安装在池塘不同深度和位置，通过RS485总线、LoRa无线模块或4G DTU把数据回传到网关。网关再通过局域网或互联网把数据推送到服务器。服务器端需要一个能稳定接收、解析、存储并展示数据的程序。R语言虽然在统计建模中更常见，但它的Plumber包可以把R函数暴露为HTTP接口，httr包可以主动请求传感器网关，serial包可以读取本地串口，Shiny则能快速做出网页监控面板。围绕这些能力，完全可以用R搭建一套小型智慧渔业数据传输与监控系统。
一、网络架构与数据流向
水质传感器数据链路一般分为三层：采集层、汇聚层和应用层。采集层由溶解氧电极、pH电极、温度探头等组成，很多传感器输出RS485信号，通过Modbus RTU协议与现场网关通信。汇聚层可以是树莓派、工控机或带边缘计算功能的DTU，它负责把串口数据转成TCP、HTTP或MQTT报文。应用层则接收这些报文，完成入库、分析和报警。R语言主要工作在汇聚层和应用层，它既可以在工控机上直接读取串口，也可以跑在服务器上接收网关转发的数据。
小型养殖场不必追求复杂微服务。一条链路可以简化为：传感器节点每分钟采集一次数据，网关用HTTP POST发送到R服务的Plumber接口，接口将数据写入SQLite或InfluxDB，Shiny应用定时读取最新记录并刷新曲线。这种架构部署成本低，排错路径短，适合几十个到几百个节点的塘口规模。
需要特别注意的是，水质传感器数据不是孤立的数值，必须带上时间戳、池塘编号和传感器位置。否则后期分析时无法判断同一池塘不同深度的溶解氧差异。因此数据报文至少应包含pond_id、sensor_id、value、unit和timestamp五个字段，这一点在网关转发时就要约定好。
二、用R接入传感器数据的常见方法
第一种是直接读串口。如果R程序运行在现场工控机上，可以用serial包打开COM口或TTY设备。下面代码演示读取一个以文本行输出的溶解氧传感器：
library(serial) con <- serialConnection( name = "do_sensor", port = "COM3", mode = "9600,n,8,1" ) open(con) Sys.sleep(2) line <- read.serialConnection(con) close(con) cat(line)
如果传感器走Modbus RTU，直接逐字节解析比较麻烦。更稳妥的做法是让网关完成Modbus解析，R只通过HTTP或MQTT接收结构化的JSON或表单数据。第二种方式是用Plumber发布HTTP接口，网关使用POST请求把数据推过来。下面是一个接收水质数据的Plumber接口：
library(plumber)
library(DBI)
library(RSQLite)
con <- dbConnect(SQLite(), "water_quality.db")
#* @post /api/sensor
#* @param pond_id 池塘编号
#* @param sensor_id 传感器编号
#* @param value 测量值
#* @param unit 单位
function(pond_id, sensor_id, value, unit) {
record <- data.frame(
time = Sys.time(),
pond_id = as.character(pond_id),
sensor_id = as.character(sensor_id),
value = as.numeric(value),
unit = as.character(unit)
)
dbAppendTable(con, "water_quality", record)
list(status = "ok", time = format(Sys.time(), "%Y-%m-%d %H:%M:%S"))
}
启动服务时执行pr <- plumber::plumb("api.R"); pr$run(host = "0.0.0.0", port = 8000)。网关或测试脚本可以用httr包模拟发送：
library(httr)
res <- POST(
url = "http://127.0.0.1:8000/api/sensor",
body = list(
pond_id = "P01",
sensor_id = "DO01",
value = 6.8,
unit = "mg/L"
),
encode = "form"
)
content(res)
第三种是MQTT订阅。对于无线传感器网络，MQTT比HTTP更省带宽，也支持离线消息。R中可以使用mqtt包连接到Broker，监听塘口主题。网关把数据发布到pond/P01/DO这样的主题，R服务端收到后立即入库。MQTT更适合需要双向控制和多订阅端的场景，例如既要存储数据，又要推送报警到手机。
三、数据清洗与异常检测不能省
传感器在池塘中长期浸泡，电极表面容易附着藻类或气泡，数据会出现毛刺、漂移和短时中断。直接把这些原始值画成曲线，经常会出现断崖式跳变，误导养殖决策。因此在入库前或入库后需要做清洗。
常用的清洗策略包括：设置物理合理范围，例如溶解氧0到20 mg/L，pH 0到14，水温-2到45摄氏度；对单点突变进行滑动窗口中位数替换；对缺失值采用线性插值或上一有效值填充。下面用dplyr和zoo包完成一次滑动平滑和异常标记：
library(dplyr)
library(zoo)
df <- df %>%
arrange(time) %>%
mutate(
do_range_ok = do >= 0 & do <= 20,
do_smooth = rollmedian(do, k = 5, fill = NA, align = "center"),
do_anomaly = abs(do - do_smooth) > 1.5
)
如果某个池塘的溶解氧在五分钟内从6 mg/L降到2 mg/L，而其他传感器没有类似变化，应该先检查传感器是否被污泥覆盖，而不是立刻开启增氧机。R脚本可以设置双重确认逻辑：连续三个采样点低于阈值才触发报警，避免误报。
另外，数据入库时建议同时保留原始值和清洗后的标志位。这样后续分析可以追溯哪些点被算法修正过，也便于根据现场实际情况调整阈值。
四、Shiny看板与报警机制
数据进入SQLite或InfluxDB后，如果只能通过命令行查询，使用门槛太高。Shiny可以把水质变化直观展示出来。下面是一个简单看板，包含池塘选择器和溶解氧趋势图：
library(shiny)
library(ggplot2)
ui <- fluidPage(
titlePanel("智慧渔业水质监控"),
sidebarLayout(
sidebarPanel(
selectInput("pond", "池塘编号", choices = c("P01", "P02", "P03"))
),
mainPanel(
plotOutput("trend")
)
)
)
server <- function(input, output) {
output$trend <- renderPlot({
df %>%
filter(pond_id == input$pond) %>%
ggplot(aes(time, do)) +
geom_line(color = "#0072B5") +
labs(
title = paste(input$pond, "溶解氧变化"),
x = "时间",
y = "溶解氧(mg/L)"
)
})
}
shinyApp(ui, server)
报警可以集成在看板中，也可以独立运行一个定时脚本。例如每五分钟汇总最新数据，若溶解氧低于4 mg/L且持续10分钟，就调用邮件或企业微信接口发送消息。R中可以用blastula包发邮件，或者用httr请求企业微信机器人的Webhook。报警内容应包含池塘编号、当前值、触发阈值和建议措施，这样值班人员不用再查曲线就能快速行动。
五、部署稳定性与性能优化
R服务在长时间运行时要考虑连接管理。Plumber接口每次请求都会创建新的数据库连接吗？如果连接写在函数外部，启动时创建一次，后续共享即可。SQLite写入并发不高时问题不大，但多个网关同时上报时应使用WAL模式降低锁冲突。
如果数据量增长较快，建议把历史数据迁移到时序数据库，例如InfluxDB或TimescaleDB。R可以通过influxdbr或DBI连接这些数据库，保留近30天明细，超过30天的数据按小时聚合。这样Shiny看板加载最近一周曲线时不会扫描全表。
对于传感器掉线和网络抖动，还需要在网关侧做本地缓存。例如树莓派网关临时把数据写入本地CSV，网络恢复后再补传。R服务端可以根据时间戳去重，避免重复入库。补传数据到达时，先按pond_id、sensor_id和time查询是否存在，存在则跳过。这样即使互联网中断几个小时，也不会丢失池塘的完整水质记录。