企业官网建设流程全解析

船舶淡水柜液位统计与补水提醒系统 —— 基于 OOP 的消耗速率预估实战

"出海最怕的不是机器坏，是生活用水见底。远洋航行动辄几十天，淡水柜容量有限，补一次水要么靠港口要么靠造水机——造水机还得烧锅炉，费油。所以你必须知道：按现在的消耗速度，这缸水还能撑几天？这不是拍脑袋能算准的。"

—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸

一、实际应用场景描述

在远洋船舶、科考船、海上平台等独立运行的生活保障系统中，淡水供应是最基础的生存要素。典型的船舶淡水系统结构如下：

┌──────────────────────────────────────────────┐

│ 船舶淡水保障系统 │

│ │

│ 舷外海水 ──→ [造水机/蒸馏装置] ──→ 淡水柜 │

│ │ │

│ ▼ │

│ ┌─────────────────────────────────────────┐ │

│ │ 淡水柜 (Capacity: 20m³) │ │

│ │ │ │

│ │ 当前液位: ████████░░░░ 62% (12.4 m³) │ │

│ │ 消耗速率: ~0.48 m³/h │ │

│ │ 预计耗尽: 25.8 小时后 │ │

│ │ 补给建议: 18h 后启动造水机 │ │

│ └─────────────────────────────────────────┘ │

│ │ ▲ │

│ ▼ │ │

│ [日用泵] → 厨房/浴室/洗涤 ──→ 废水 │

│ │ │

│ [液位变送器 4-20mA] ← 电容式/雷达式 │

└──────────────────────────────────────────────┘

消耗速率的影响因素

因素 影响方向 典型变化

船员人数 正相关 15~30人，波动不大

天气温度 高温↑消耗 热带海域 +20%

靠港/锚泊 锚泊↓消耗 靠港期间少洗澡

设备冲洗 突发↑ 甲板冲洗 +0.5m³/h

造水机状态 补充源 运行中抵消消耗

哈尔滨工程大学《工业过程控制》课程在第五章"过程建模与参数估计"中详细讲解了如何从输入输出数据中辨识过程模型：

"对于慢时变过程，可以通过滑动窗口的最小二乘法在线估计过程参数。这种递推辨识方法不需要存储全部历史数据，适合嵌入式系统和实时应用。"

二、引入痛点

2.1 现场的真实困境

场景 现场发生了什么 根因

突然断水 "早上起来水龙头没水了，全船人没法洗漱" 没预估消耗趋势，只看瞬时液位

造水机空转 "液位还很高就开了造水机，白白烧了一天油" 缺少提前量规划

补给时机难定 "到底什么时候该启动造水机？" 消耗速率是变化的，不是固定值

数据沉睡 "液位记录仪每天都在记，但没人看趋势" 缺乏自动分析工具

报警太迟 "低液位报警响了，但已经来不及造水" 报警阈值是固定的，不考虑消耗速度

2.2 核心矛盾

固定阈值的低液位报警是"马后炮"——它告诉你"水快没了"，但不告诉你"还有多久会没"。如果消耗速率是 0.3m³/h，还剩 2m³ 时报警还来得及；但如果突然有人冲洗甲板，消耗飙升到 1.2m³/h，同样剩 2m³，你只有 100 分钟了。你需要的是一个"倒计时"，而不是一个"刻度尺"。

2.3 我们要解决什么

用一段 Python 程序，构建一个船舶淡水柜液位统计与补水提醒系统，实现：

1. 液位数据加载 —— 从 CSV 读取时序液位和消耗记录

2. 消耗速率预估 —— 滑动窗口线性回归 + 指数平滑

3. 剩余时间预测 —— 基于当前速率推算耗尽时间

4. 智能补水提醒 —— 多级预警 + 考虑造水机启动延迟

5. 趋势可视化 —— 液位曲线 + 速率变化 + 预警线

6. 面向对象设计 —— 分层清晰，可扩展

三、核心逻辑讲解

3.1 理论基础：递推最小二乘与指数平滑

本工具基于哈工程《工业过程控制》第五章"过程参数估计"：

① 滑动窗口线性回归（消耗速率估计）

对最近 N 个时间点的液位数据 (t_i, L_i) ，拟合直线 L = a - r \cdot t ：

r = \frac{N\sum(t_i L_i) - \sum t_i \sum L_i}{N\sum(t_i^2) - (\sum t_i)^2}

其中 r > 0 即为消耗速率（单位：m³/h）。

② 指数平滑（消除短期波动）

r_t = \alpha \cdot r_{raw} + (1-\alpha) \cdot r_{t-1}

\alpha = 0.3 表示更相信近期测量。

③ 剩余时间预估

T_{remain} = \frac{L_{current} - L_{reserve}}{r}

其中 L_{reserve} 是安全余量（如 1.0m³，防止抽空损坏水泵）。

④ 多级预警逻辑

级别 条件 动作

INFO T_{remain} < 48h 提示关注

WARNING T_{remain} < 24h 建议准备造水

CRITICAL T_{remain} < 8h 立即启动造水机

3.2 系统数据流

┌──────────────────────────────┐

│ 液位数据 CSV │

│ (timestamp, level_m3) │

└──────────────┬───────────────┘

│

┌──────────────▼───────────────┐

│ ① 数据加载 & 滤波 │

│ 剔除跳变、填充缺失 │

└──────────────┬───────────────┘

│

┌──────────────▼───────────────┐

│ ② 消耗速率计算 │

│ 滑动窗口线性回归 │

│ + 指数平滑 │

└──────────────┬───────────────┘

│

┌──────────────▼───────────────┐

│ ③ 剩余时间预测 │

│ T = (L - L_reserve) / r │

└──────────────┬───────────────┘

│

┌──────────────▼───────────────┐

│ ④ 预警判定 │

│ 多级阈值比较 │

└──────────────┬───────────────┘

│

┌──────────────▼───────────────┐

│ ⑤ 补水建议生成 │

│ 考虑造水机延迟 │

└──────────────┬───────────────┘

│

┌──────────────▼───────────────┐

│ ⑥ 报告 & 可视化 │

└──────────────────────────────┘

四、代码讲解（面向对象设计）

4.1 类结构总览

类名 职责 设计模式

"LevelRecord" 单条液位记录（dataclass） 值对象

"TankConfig" 淡水柜参数配置（值对象） 值对象

"AlertConfig" 预警阈值配置（值对象） 值对象

"SmoothingConfig" 平滑算法配置（值对象） 值对象

"DataLoader" CSV 数据加载与滤波 封装

"ConsumptionEstimator" 消耗速率预估器 策略模式

"TimeToEmptyPredictor" 剩余时间预测器 封装

"AlertEngine" 多级预警引擎 状态模式

"RefillAdvisor" 补水建议生成器 封装

"ReportGenerator" 分析报告生成器 模板方法

"FreshwaterMonitorSystem" 系统编排器（聚合根） 聚合根

4.2 数据模型层

from dataclasses import dataclass, field

from typing import List, Optional, Tuple

from enum import Enum, auto

import numpy as np

import csv

from pathlib import Path

from collections import deque

class AlertLevel(Enum):

"""预警级别"""

NORMAL = "正常"

INFO = "提示"

WARNING = "警告"

CRITICAL = "紧急"

@dataclass(frozen=True)

class LevelRecord:

"""单条液位记录 —— 值对象"""

timestamp: float # 时间戳 (小时, 从起始计)

level: float # 液位 (m³)

source: str = "sensor" # 数据来源

@dataclass(frozen=True)

class TankConfig:

"""淡水柜配置"""

capacity: float = 20.0 # 总容量 (m³)

reserve: float = 1.0 # 安全余量 (m³)

low_level: float = 3.0 # 低液位报警线 (m³)

max_fill_time: float = 4.0 # 造水机最大启动延迟 (h)

@dataclass(frozen=True)

class AlertConfig:

"""预警阈值配置"""

info_hours: float = 48.0 # 提示: 剩余 < 48h

warning_hours: float = 24.0 # 警告: 剩余 < 24h

critical_hours: float = 8.0 # 紧急: 剩余 < 8h

@dataclass(frozen=True)

class SmoothingConfig:

"""平滑配置"""

window_size: int = 12 # 滑动窗口大小 (12个采样点)

alpha: float = 0.3 # 指数平滑系数

min_window: int = 4 # 最小有效窗口

4.3 数据加载器

class DataLoader:

"""

液位数据加载器

CSV 格式:

timestamp,level_m3,source

0.0,18.5,sensor

0.5,18.2,sensor

1.0,17.9,sensor

...

时间戳单位为小时，便于直接计算消耗速率 (m³/h)

"""

def __init__(self):

self.records: List[LevelRecord] = []

def load_csv(self, file_path: str) -> List[LevelRecord]:

"""从 CSV 加载数据"""

self.records.clear()

with open(file_path, 'r', encoding='utf-8') as f:

reader = csv.DictReader(f)

for row in reader:

record = LevelRecord(

timestamp=float(row['timestamp']),

level=float(row['level_m3']),

source=row.get('source', 'sensor')

)

self.records.append(record)

return self.records

def median_filter(self, window: int = 3) -> List[LevelRecord]:

"""

中值滤波 —— 消除传感器偶发跳变

对液位数据做滑动中值滤波，保留趋势但去除尖峰

"""

if len(self.records) < window:

return self.records

filtered = []

values = [r.level for r in self.records]

for i in range(len(self.records)):

start = max(0, i - window // 2)

end = min(len(values), i + window // 2 + 1)

median_val = float(np.median(values[start:end]))

filtered.append(LevelRecord(

timestamp=self.records[i].timestamp,

level=median_val,

source=self.records[i].source

))

return filtered

4.4 消耗速率预估器（核心算法）

class ConsumptionEstimator:

"""

消耗速率预估器

方法:

1. 滑动窗口线性回归: 计算局部消耗斜率

2. 指数平滑: 融合历史估计，减少抖动

物理意义:

液位随时间下降的斜率就是消耗速率

rate = -dL/dt

"""

def __init__(self, smoothing_cfg: SmoothingConfig):

self.cfg = smoothing_cfg

self._smoothed_rate: Optional[float] = None

self._rate_history: deque = deque(maxlen=100)

def compute_raw_rate(self, records: List[LevelRecord]) -> float:

"""

基于两点计算瞬时消耗速率

Args:

records: 两条记录 [较早, 较新]

Returns:

消耗速率 (m³/h), 正值表示消耗

"""

if len(records) < 2:

return 0.0

dt = records[-1].timestamp - records[0].timestamp

if dt <= 0:

return 0.0

dl = records[0].level - records[-1].level # 液位下降量

return max(0.0, dl / dt) # 只关心消耗(正值)

def sliding_window_regression(self, records: List[LevelRecord]) -> float:

"""

滑动窗口线性回归计算消耗速率

对窗口内的 (t, L) 拟合直线 L = a + b*t

消耗速率 = -b (取正值)

Returns:

消耗速率 (m³/h)

"""

n = len(records)

if n < self.cfg.min_window:

return 0.0

# 取最近的 window_size 条记录

window = records[-self.cfg.window_size:] if n > self.cfg.window_size else records

n_win = len(window)

t = np.array([r.timestamp for r in window])

L = np.array([r.level for r in window])

# 线性回归: L = a + b*t

t_mean = np.mean(t)

L_mean = np.mean(L)

numerator = np.sum((t - t_mean) * (L - L_mean))

denominator = np.sum((t - t_mean) ** 2)

if denominator == 0:

return 0.0

b = numerator / denominator # 斜率

rate = max(0.0, -b) # 消耗速率 = -斜率

return float(rate)

def exponential_smooth(self, raw_rate: float) -> float:

"""

指数平滑

r_t = α * r_raw + (1-α) * r_{t-1}

Args:

raw_rate: 原始速率估计

Returns:

平滑后的速率

"""

if self._smoothed_rate is None:

self._smoothed_rate = raw_rate

else:

self._smoothed_rate = (self.cfg.alpha * raw_rate +

(1 - self.cfg.alpha) * self._smoothed_rate)

self._rate_history.append(self._smoothed_rate)

return self._smoothed_rate

def get_current_rate(self, records: List[LevelRecord]) -> float:

"""

获取当前消耗速率 (完整流程)

Returns:

平滑后的消耗速率 (m³/h)

"""

raw = self.sliding_window_regression(records)

smoothed = self.exponential_smooth(raw)

return smoothed

def reset(self):

"""重置状态"""

self._smoothed_rate = None

self._rate_history.clear()

4.5 剩余时间预测器

class TimeToEmptyPredictor:

"""

剩余时间预测器

计算: T_remain = (L_current - L_reserve) / consumption_rate

"""

def __init__(self, tank_cfg: TankConfig):

self.tank_cfg = tank_cfg

def predict(self, current_level: float, rate: float) -> dict:

"""

预测剩余可用时间

Args:

current_level: 当前液位 (m³)

rate: 消耗速率 (m³/h)

Returns:

预测结果字典

"""

available = current_level - self.tank_cfg.reserve

if available <= 0:

return {

'hours_remaining': 0.0,

'minutes_remaining': 0.0,

'available_volume': 0.0,

'rate': rate,

'exhausted': True

}

if rate <= 0.001:

# 速率极低，视为无限

return {

'hours_remaining': float('inf'),

'minutes_remaining': float('inf'),

'available_volume': round(available, 2),

'rate': rate,

'exhausted': False

}

hours = available / rate

return {

'hours_remaining': round(hours, 2),

'minutes_remaining': round(hours * 60, 1),

'available_volume': round(available, 2),

'rate': round(rate, 3),

'exhausted': False

}

4.6 多级预警引擎

class AlertEngine:

"""

多级预警引擎 —— 状态模式

根据剩余时间判定预警级别:

- NORMAL: > 48h

- INFO: 24h ~ 48h

- WARNING: 8h ~ 24h

- CRITICAL: < 8h

"""

def __init__(self, alert_cfg: AlertConfig):

self.cfg = alert_cfg

self.current_level = AlertLevel.NORMAL

self.history: List[Tuple[float, AlertLevel, str]] = []

def evaluate(self, hours_remaining: float, timestamp: float = 0.0) -> AlertLevel:

"""

评估预警级别

Args:

hours_remaining: 剩余可用时间 (h)

timestamp: 当前时间戳

Returns:

预警级别

"""

if hours_remaining == float('inf'):

self.current_level = AlertLevel.NORMAL

elif hours_remaining < self.cfg.critical_hours:

self.current_level = AlertLevel.CRITICAL

elif hours_remaining < self.cfg.warning_hours:

self.current_level = AlertLevel.WARNING

elif hours_remaining < self.cfg.info_hours:

self.current_level = AlertLevel.INFO

else:

self.current_level = AlertLevel.NORMAL

# 记录状态变化

message = self._get_message(self.current_level, hours_remaining)

self.history.append((timestamp, self.current_level, message))

return self.current_level

def _get_message(self, level: AlertLevel, hours: float) -> str:

"""生成预警消息"""

if level == AlertLevel.CRITICAL:

return f"⚠️ 紧急: 淡水预计 {hours:.1f}h 后耗尽! 立即启动造水机!"

elif level == AlertLevel.WARNING:

return f"⚠️ 警告: 淡水预计 {hours:.1f}h 后耗尽, 请准备启动造水机"

elif level == AlertLevel.INFO:

return f"ℹ️ 提示: 淡水预计 {hours:.1f}h 后耗尽, 建议关注消耗情况"

else:

return f"✓ 淡水充足, 剩余 {hours:.1f}h"

def get_status_summary(self) -> str:

"""获取状态摘要"""

if not self.history:

return "无预警记录"

latest = self.history[-1]

return f"[{latest[1].value}] {latest[2]}"

4.7 补水建议生成器

class RefillAdvisor:

"""

补水建议生成器

综合考虑:

1. 当前消耗速率

2. 造水机启动延迟

3. 安全余量

4. 预计补给到达时间

"""

def __init__(self, tank_cfg: TankConfig, alert_cfg: AlertConfig):

self.tank_cfg = tank_cfg

self.alert_cfg = alert_cfg

def advise(self, prediction: dict, rate: float) -> dict:

"""

生成补水建议

Args:

prediction: TimeToEmptyPredictor.predict() 的结果

rate: 当前消耗速率

Returns:

建议字典

"""

hours_left = prediction.get('hours_remaining', float('inf'))

if hours_left == float('inf') or hours_left > self.alert_cfg.info_hours:

return {

'action': '无需操作',

'start_makeup_at': None,

'priority': 'low',

'message': '淡水充足，维持正常监控'

}

# 造水机需要提前启动的时间

lead_time = self.tank_cfg.max_fill_time

if hours_left <= self.alert_cfg.critical_hours:

return {

'action': '立即启动造水机',

'start_makeup_at': 'NOW',

'priority': 'critical',

'message': f'剩余 {hours_left:.1f}h，已超过造水机启动延迟({lead_time}h)，立即行动!'

}

elif hours_left <= self.alert_cfg.warning_hours:

return {

'action': '计划启动造水机',

'start_makeup_at': f'{hours_left - lead_time:.1f}h 后',

'priority': 'high',

'message': f'建议在 {hours_left - lead_time:.1f}h 内启动造水机'

}

else:

return {

'action': '监控消耗趋势',

'start_makeup_at': f'{hours_left - lead_time:.1f}h 后',

'priority': 'medium',

'message': f'当前速率 {rate:.3f} m³/h, 预计 {hours_left:.1f}h 后需补水'

}

4.8 分析报告生成器

class ReportGenerator:

"""

分析报告生成器

"""

def generate(self, tank_cfg: TankConfig, current_level: float,

rate: float, prediction: dict, alert_status: str,

advice: dict) -> str:

"""生成文本报告"""

lines = [

"=" * 60,

" 船舶淡水柜液位监测与补水提醒报告",

"=" * 60,

"",

" 【当前状态】",

f" 淡水柜容量: {tank_cfg.capacity} m³",

f" 当前液位: {current_level:.2f} m³ ({current_level/tank_cfg.capacity*100:.1f}%)",

f" 安全余量: {tank_cfg.reserve} m³",

f" 可用水量: {prediction.get('available_volume', 0):.2f} m³",

"",

" 【消耗分析】",

f" 当前消耗速率: {rate:.3f} m³/h",

f" 日消耗量: {rate * 24:.1f} m³/day",

f" 人均估算: {rate * 24 / 20:.1f} L/人/天 (按20人计)",

"",

" 【剩余时间预测】",

f" 预计耗尽时间: {prediction.get('hours_remaining', '∞'):.1f} h",

f" 即约 {prediction.get('minutes_remaining', '∞'):.0f} 分钟后",

"",

" 【预警状态】",

f" {alert_status}",

"",

" 【补水建议】",

f" 建议动作: {advice.get('action', 'N/A')}",

f" 优先级: {advice.get('priority', 'N/A')}",

f" 详情: {advice.get('message', 'N/A')}",

"",

"=" * 60

]

return "\n".join(lines)

4.9 系统编排器

class FreshwaterMonitorSystem:

"""

船舶淡水监测系统 —— 聚合根

串联: 加载 → 滤波 → 速率预估 → 预测 → 预警 → 建议 → 报告

"""

def __init__(self, tank_cfg: TankConfig = None,

alert_cfg: AlertConfig = None,

smoothing_cfg: SmoothingConfig = None):

self.tank_cfg = tank_cfg or TankConfig()

self.alert_cfg = alert_cfg or AlertConfig()

self.smoothing_cfg = smoothing_cfg or SmoothingConfig()

self.loader = DataLoader()

self.estimator = ConsumptionEstimator(self.smoothing_cfg)

self.predictor = TimeToEmptyPredictor(self.tank_cfg)

self.alert_engine = AlertEngine(self.alert_cfg)

self.advisor = RefillAdvisor(self.tank_cfg, self.alert_cfg)

self.reporter = ReportGenerator()

def analyze(self, data_path: str) -> str:

"""

执行完整分析流程

Args:

data_path: 液位数据 CSV 路径

Returns:

分析报告

"""

# ① 加载

records = self.loader.load_csv(data_path)

if len(records) < 2:

return "错误: 数据不足，至少需要 2 条记录"

# ② 滤波

filtered = self.loader.median_filter(window=3)

# ③ 速率预估

rate = self.estimator.get_current_rate(filtered)

current_level = filtered[-1].level

# ④ 剩余时间预测

prediction = self.predictor.predict(current_level, rate)

# ⑤ 预警评估

hours = prediction.get('hours_remaining', float('inf'))

timestamp = filtered[-1].timestamp

alert_level = self.alert_engine.evaluate(hours, timestamp)

alert_status = self.alert_engine.get_status_summary()

# ⑥ 补水建议

advice = self.advisor.advise(prediction, rate)

# ⑦ 生成报告

report = self.reporter.generate(

self.tank_cfg, current_level, rate,

prediction, alert_status, advice

)

return report

4.10 完整演示

def demo():

"""完整演示"""

print("=" * 60)

print(" 船舶淡水柜液位统计与补水提醒系统 v1.0")

print(" 基于哈尔滨工程大学《工业过程控制》课程理论")

print("=" * 60)

import tempfile

import os

# 生成模拟液位数据 (模拟48小时, 每30分钟一条)

np.random.seed(42)

csv_lines = ["timestamp,level_m3,source"]

# 初始液位 18.5 m³, 消耗速率约 0.38 m³/h

# 第24h后消耗加快到 0.55 m³/h (天气热, 洗澡多了)

current_level = 18.5

for i in range(97): # 0~48h, 每0.5h

t = i * 0.5

if t < 24:

rate = 0.38 + np.random.normal(0, 0.02)

else:

rate = 0.55 + np.random.normal(0, 0.03)

current_level -= rate * 0.5

current_level = max(0.5, current_level) # 不低于0.5

csv_lines.append(f"{t:.1f},{current_level:.3f},sensor")

# 写入临时文件

tmpdir = tempfile.mkdtemp()

csv_path = os.path.join(tmpdir, "level_data.csv")

with open(csv_path, 'w') as f:

f.write("\n".join(csv_lines))

# 运行分析

system = FreshwaterMonitorSystem(

tank_cfg=TankConfig(capacity=20.0, reserve=1.0),

alert_cfg=AlertConfig(info_hours=48.0, warning_hours=24.0, critical_hours=8.0),

s

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！