船舶淡水柜液位统计与补水提醒系统 —— 基于 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, fieldfrom typing import List, Optional, Tuplefrom enum import Enum, autoimport numpy as npimport csvfrom pathlib import Pathfrom collections import dequeclass AlertLevel(Enum):预警级别NORMAL 正常INFO 提示WARNING 警告CRITICAL 紧急dataclass(frozenTrue)class LevelRecord:单条液位记录 —— 值对象timestamp: float # 时间戳 (小时, 从起始计)level: float # 液位 (m³)source: str sensor # 数据来源dataclass(frozenTrue)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(frozenTrue)class AlertConfig:预警阈值配置info_hours: float 48.0 # 提示: 剩余 48hwarning_hours: float 24.0 # 警告: 剩余 24hcritical_hours: float 8.0 # 紧急: 剩余 8hdataclass(frozenTrue)class SmoothingConfig:平滑配置window_size: int 12 # 滑动窗口大小 (12个采样点)alpha: float 0.3 # 指数平滑系数min_window: int 4 # 最小有效窗口4.3 数据加载器class DataLoader:液位数据加载器CSV 格式:timestamp,level_m3,source0.0,18.5,sensor0.5,18.2,sensor1.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, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:record LevelRecord(timestampfloat(row[timestamp]),levelfloat(row[level_m3]),sourcerow.get(source, sensor))self.records.append(record)return self.recordsdef median_filter(self, window: int 3) - List[LevelRecord]:中值滤波 —— 消除传感器偶发跳变对液位数据做滑动中值滤波保留趋势但去除尖峰if len(self.records) window:return self.recordsfiltered []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(timestampself.records[i].timestamp,levelmedian_val,sourceself.records[i].source))return filtered4.4 消耗速率预估器核心算法class ConsumptionEstimator:消耗速率预估器方法:1. 滑动窗口线性回归: 计算局部消耗斜率2. 指数平滑: 融合历史估计减少抖动物理意义:液位随时间下降的斜率就是消耗速率rate -dL/dtdef __init__(self, smoothing_cfg: SmoothingConfig):self.cfg smoothing_cfgself._smoothed_rate: Optional[float] Noneself._rate_history: deque deque(maxlen100)def compute_raw_rate(self, records: List[LevelRecord]) - float:基于两点计算瞬时消耗速率Args:records: 两条记录 [较早, 较新]Returns:消耗速率 (m³/h), 正值表示消耗if len(records) 2:return 0.0dt records[-1].timestamp - records[0].timestampif dt 0:return 0.0dl 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 recordsn_win len(window)t np.array([r.timestamp for r in window])L np.array([r.level for r in window])# 线性回归: L a b*tt_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.0b 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_rateelse: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_ratedef get_current_rate(self, records: List[LevelRecord]) - float:获取当前消耗速率 (完整流程)Returns:平滑后的消耗速率 (m³/h)raw self.sliding_window_regression(records)smoothed self.exponential_smooth(raw)return smootheddef reset(self):重置状态self._smoothed_rate Noneself._rate_history.clear()4.5 剩余时间预测器class TimeToEmptyPredictor:剩余时间预测器计算: T_remain (L_current - L_reserve) / consumption_ratedef __init__(self, tank_cfg: TankConfig):self.tank_cfg tank_cfgdef predict(self, current_level: float, rate: float) - dict:预测剩余可用时间Args:current_level: 当前液位 (m³)rate: 消耗速率 (m³/h)Returns:预测结果字典available current_level - self.tank_cfg.reserveif 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 / ratereturn {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: 8hdef __init__(self, alert_cfg: AlertConfig):self.cfg alert_cfgself.current_level AlertLevel.NORMALself.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.NORMALelif hours_remaining self.cfg.critical_hours:self.current_level AlertLevel.CRITICALelif hours_remaining self.cfg.warning_hours:self.current_level AlertLevel.WARNINGelif hours_remaining self.cfg.info_hours:self.current_level AlertLevel.INFOelse: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_leveldef _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}hdef 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_cfgself.alert_cfg alert_cfgdef 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_timeif 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(window3)# ③ 速率预估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].timestampalert_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 report4.10 完整演示def demo():完整演示print( * 60)print( 船舶淡水柜液位统计与补水提醒系统 v1.0)print( 基于哈尔滨工程大学《工业过程控制》课程理论)print( * 60)import tempfileimport 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.5for i in range(97): # 0~48h, 每0.5ht i * 0.5if 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.5current_level max(0.5, current_level) # 不低于0.5csv_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_cfgTankConfig(capacity20.0, reserve1.0),alert_cfgAlertConfig(info_hours48.0, warning_hours24.0, critical_hours8.0),s利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
