智能饮水管理程序设计与实现
一、实际应用场景与痛点分析
应用场景
现代人工作繁忙，常常忽视科学饮水，导致身体脱水、新陈代谢下降、注意力不集中等问题。本程序面向需要科学管理饮水的用户，特别是办公室人员、运动员、户外工作者和特殊健康状况人群。
主要痛点
1. 忘记饮水 - 工作繁忙时常忘记喝水
2. 饮水不足 - 不了解自身实际需水量
3. 饮水时机不当 - 集中在某个时间段饮水
4. 忽略环境因素 - 气温、湿度、运动量等影响
5. 缺乏个性化 - 标准建议不适应个体差异
6. 难以坚持 - 缺乏提醒和激励
二、核心逻辑与智能控制原理
系统架构
数据感知层 → 数据处理层 → 智能决策层 → 反馈执行层
↓ ↓ ↓ ↓
环境监测 → 需水计算 → 饮水决策 → 智能提醒
用户输入 → 模糊推理 → 时间优化 → 效果评估
核心智能控制原理
1. 模糊控制 - 处理"高温"、"中强度运动"等模糊概念
2. 专家系统 - 基于医学和运动科学知识的规则库
3. 预测控制 - 基于生理模型预测需水量
4. 自适应控制 - 根据用户反馈调整计划
5. 强化学习 - 优化提醒时机和方式
三、代码实现
主程序：smart_hydration_manager.py
#!/usr/bin/env python3
"""
智能饮水管理系统
基于智能控制原理的个性化饮水计划与提醒系统
"""
import json
import datetime
import time
import numpy as np
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
import matplotlib.pyplot as plt
from matplotlib.patches import Circle
import threading
import warnings
from abc import ABC, abstractmethod
import random
import logging
from dataclasses_json import dataclass_json
import os
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('hydration.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class ActivityLevel(Enum):
"""活动水平枚举（模糊集合）"""
SEDENTARY = 1.0 # 久坐
LIGHT = 1.2 # 轻度活动
MODERATE = 1.5 # 中度活动
HIGH = 1.8 # 高强度
EXTREME = 2.2 # 极端强度
class EnvironmentFactor(Enum):
"""环境因素枚举"""
TEMPERATURE = "temperature" # 温度
HUMIDITY = "humidity" # 湿度
ALTITUDE = "altitude" # 海拔
INDOOR = "indoor" # 室内外
class WaterSource(Enum):
"""水源类型枚举"""
PLAIN_WATER = "白水"
TEA = "茶"
COFFEE = "咖啡"
MILK = "牛奶"
JUICE = "果汁"
SPORTS_DRINK = "运动饮料"
@dataclass_json
@dataclass
class WaterIntake:
"""饮水记录"""
timestamp: datetime.datetime
amount: float # 毫升
source: WaterSource
temperature: float = 20.0 # 水温，摄氏度
note: str = "" # 备注
@dataclass_json
@dataclass
class SweatRecord:
"""出汗记录"""
timestamp: datetime.datetime
amount: float # 毫升
activity_type: str
duration: float # 分钟
intensity: float # 强度 0-1
@dataclass_json
@dataclass
class EnvironmentData:
"""环境数据"""
timestamp: datetime.datetime
temperature: float # 摄氏度
humidity: float # 百分比
altitude: float = 0 # 米
is_indoor: bool = True
@dataclass_json
@dataclass
class UserProfile:
"""用户基本信息"""
user_id: str
name: str
age: int
gender: str # 'male' or 'female'
weight: float # 公斤
height: float # 厘米
activity_level: ActivityLevel
health_conditions: List[str] # 健康状况
wake_up_time: datetime.time = datetime.time(7, 0) # 起床时间
sleep_time: datetime.time = datetime.time(23, 0) # 睡觉时间
preferences: Dict[str, Any] = None # 用户偏好
class FuzzyHydrationController:
"""
模糊饮水控制器
处理温度、湿度、活动强度等模糊概念
"""
def __init__(self):
self.temperature_rules = self._init_temperature_rules()
self.humidity_rules = self._init_humidity_rules()
self.activity_rules = self._init_activity_rules()
def _init_temperature_rules(self) -> Dict[str, Dict]:
"""初始化温度模糊规则"""
return {
"cold": {"min": -10, "optimal": 15, "max": 20, "weight": 0.9},
"cool": {"min": 15, "optimal": 22, "max": 25, "weight": 1.0},
"warm": {"min": 22, "optimal": 28, "max": 30, "weight": 1.2},
"hot": {"min": 28, "optimal": 33, "max": 35, "weight": 1.5},
"very_hot": {"min": 33, "optimal": 38, "max": 45, "weight": 2.0}
}
def _init_humidity_rules(self) -> Dict[str, Dict]:
"""初始化湿度模糊规则"""
return {
"dry": {"min": 0, "optimal": 30, "max": 40, "weight": 1.3},
"comfortable": {"min": 30, "optimal": 50, "max": 60, "weight": 1.0},
"humid": {"min": 50, "optimal": 70, "max": 80, "weight": 0.9},
"very_humid": {"min": 70, "optimal": 85, "max": 100, "weight": 0.8}
}
def _init_activity_rules(self) -> Dict[str, Dict]:
"""初始化活动强度模糊规则"""
return {
"sedentary": {"min": 0, "optimal": 0.2, "max": 0.4, "weight": 1.0},
"light": {"min": 0.3, "optimal": 0.5, "max": 0.6, "weight": 1.3},
"moderate": {"min": 0.5, "optimal": 0.7, "max": 0.8, "weight": 1.6},
"intense": {"min": 0.7, "optimal": 0.85, "max": 0.9, "weight": 2.0},
"extreme": {"min": 0.85, "optimal": 0.95, "max": 1.0, "weight": 2.5}
}
def calculate_temperature_factor(self, temp: float) -> float:
"""计算温度对饮水需求的影响因子"""
membership = {}
for level, params in self.temperature_rules.items():
if params["min"] <= temp <= params["max"]:
# 计算隶属度（三角隶属函数）
if temp <= params["optimal"]:
membership[level] = (temp - params["min"]) / (params["optimal"] - params["min"])
else:
membership[level] = (params["max"] - temp) / (params["max"] - params["optimal"])
else:
membership[level] = 0
# 加权平均计算影响因子
total_weight = sum(membership.values())
if total_weight > 0:
factor = sum(membership[level] * self.temperature_rules[level]["weight"]
for level in membership) / total_weight
else:
factor = 1.0
return round(factor, 2)
def calculate_humidity_factor(self, humidity: float) -> float:
"""计算湿度对饮水需求的影响因子"""
membership = {}
for level, params in self.humidity_rules.items():
if params["min"] <= humidity <= params["max"]:
if humidity <= params["optimal"]:
membership[level] = (humidity - params["min"]) / (params["optimal"] - params["min"])
else:
membership[level] = (params["max"] - humidity) / (params["max"] - params["optimal"])
else:
membership[level] = 0
total_weight = sum(membership.values())
if total_weight > 0:
factor = sum(membership[level] * self.humidity_rules[level]["weight"]
for level in membership) / total_weight
else:
factor = 1.0
return round(factor, 2)
def calculate_activity_factor(self, intensity: float) -> float:
"""计算活动强度对饮水需求的影响因子"""
membership = {}
for level, params in self.activity_rules.items():
if params["min"] <= intensity <= params["max"]:
if intensity <= params["optimal"]:
membership[level] = (intensity - params["min"]) / (params["optimal"] - params["min"])
else:
membership[level] = (params["max"] - intensity) / (params["max"] - params["optimal"])
else:
membership[level] = 0
total_weight = sum(membership.values())
if total_weight > 0:
factor = sum(membership[level] * self.activity_rules[level]["weight"]
for level in membership) / total_weight
else:
factor = 1.0
return round(factor, 2)
def fuzzy_inference(self, temp: float, humidity: float, activity: float) -> Dict[str, float]:
"""模糊推理综合影响因子"""
temp_factor = self.calculate_temperature_factor(temp)
humidity_factor = self.calculate_humidity_factor(humidity)
activity_factor = self.calculate_activity_factor(activity)
# 综合影响因子（加权平均）
total_factor = (temp_factor * 0.4 +
humidity_factor * 0.3 +
activity_factor * 0.3)
return {
"temperature_factor": temp_factor,
"humidity_factor": humidity_factor,
"activity_factor": activity_factor,
"total_factor": total_factor
}
class HydrationCalculator:
"""
科学饮水计算器
基于生理学公式计算需水量
"""
@staticmethod
def calculate_basic_water_need(weight: float, activity_level: ActivityLevel) -> float:
"""
计算基础需水量
基础公式：体重(kg) × 30-40 ml
根据活动水平调整
"""
base_need = weight * 35 # 35 ml/kg
adjusted_need = base_need * activity_level.value
return adjusted_need
@staticmethod
def calculate_sweat_loss(weight_before: float, weight_after: float,
water_intake: float, urine_output: float = 0) -> float:
"""
计算出汗量（汗液流失量）
公式：Δ体重 + 饮水 - 排尿
单位：毫升
"""
weight_change = (weight_before - weight_after) * 1000 # kg to ml (近似)
sweat_loss = weight_change + water_intake - urine_output
return max(sweat_loss, 0)
@staticmethod
def calculate_hourly_water_loss(weight: float, temp: float,
humidity: float, activity: float) -> float:
"""
计算每小时水分流失
包括：基础代谢、呼吸、皮肤蒸发、出汗
"""
# 基础水分流失（不活动）
basal_loss = weight * 0.5 # ml/hour
# 温度影响
temp_factor = max(1.0, (temp - 20) * 0.05) # 每超过20度增加5%
# 湿度影响
humidity_factor = 1.0
if humidity < 30:
humidity_factor = 1.3 # 干燥增加流失
elif humidity > 70:
humidity_factor = 0.9 # 潮湿减少流失
# 活动影响
activity_loss = activity * weight * 10 # ml/hour
total_loss = basal_loss * temp_factor * humidity_factor + activity_loss
return total_loss
@staticmethod
def calculate_optimal_water_temperature(env_temp: float,
activity_level: float) -> float:
"""
计算最佳饮水温度
基于环境温度和活动水平
"""
if activity_level > 0.7: # 高强度运动
optimal_temp = 10 # 运动后适宜10-15度
elif env_temp > 30: # 高温环境
optimal_temp = 15
elif env_temp < 10: # 低温环境
optimal_temp = 40
else: # 常温环境
optimal_temp = 20
return optimal_temp
class HydrationExpertSystem:
"""
饮水专家系统
基于医学知识的规则库
"""
def __init__(self):
self.rules = self._initialize_rules()
self.water_source_effects = self._initialize_water_effects()
def _initialize_rules(self) -> List[Dict]:
"""初始化专家规则库"""
return [
{
"name": "脱水警告",
"condition": lambda data: data.get('water_balance', 0) < -500,
"action": "立即补充200-300ml水分，小口慢饮",
"severity": "high",
"message": "⚠️ 检测到轻度脱水风险，请立即饮水！"
},
{
"name": "饮水不足",
"condition": lambda data: data.get('intake_percentage', 0) < 0.5 and
data.get('hours_awake', 0) > 8,
"action": "加快饮水进度，每小时至少200ml",
"severity": "medium",
"message": "💧 今日饮水进度较慢，建议增加饮水频率"
},
{
"name": "过量饮水风险",
"condition": lambda data: data.get('hourly_intake', 0) > 1000,
"action": "暂停饮水1-2小时，让身体调节",
"severity": "medium",
"message": "⚠️ 短时间内饮水过多，可能引起水中毒"
},
{
"name": "睡前饮水提醒",
"condition": lambda data: data.get('hours_to_sleep', 0) < 2 and
data.get('recent_intake', 0) < 200,
"action": "睡前1小时饮用100-200ml温水",
"severity": "low",
"message": "🌙 睡前适量饮水，有助于夜间新陈代谢"
},
{
"name": "运动前饮水",
"condition": lambda data: data.get('planned_exercise', False) and
data.get('pre_exercise_hydration', 0) < 300,
"action": "运动前30分钟补充300-500ml水分",
"severity": "medium",
"message": "🏃 运动前请提前补充水分"
},
{
"name": "运动后补水",
"condition": lambda data: data.get('post_exercise', False) and
data.get('sweat_loss', 0) > 500,
"action": f"运动后按每损失1公斤体重补充1.5升水的原则补水",
"severity": "high",
"message": "💦 运动后需要及时补充水分和电解质"
},
{
"name": "起床饮水",
"condition": lambda data: data.get('hours_since_wakeup', 0) < 1 and
data.get('morning_intake', 0) < 300,
"action": "起床后30分钟内饮用300-500ml温水",
"severity": "medium",
"message": "🌅 晨起第一杯水，帮助身体唤醒"
},
{
"name": "餐前饮水",
"condition": lambda data: data.get('hours_since_last_meal', 0) > 3 and
data.get('pre_meal_intake', 0) < 100,
"action": "餐前30分钟饮用100-200ml水",
"severity": "low",
"message": "🍽️ 餐前适量饮水，有助于消化"
}
]
def _initialize_water_effects(self) -> Dict[str, Dict]:
"""初始化不同水源的补水效果"""
return {
"白水": {"hydration_factor": 1.0, "notes": "最佳补水选择"},
"茶": {"hydration_factor": 0.9, "notes": "含咖啡因，适量饮用"},
"咖啡": {"hydration_factor": 0.8, "notes": "利尿作用，需额外补水"},
"牛奶": {"hydration_factor": 0.95, "notes": "含蛋白质和电解质"},
"果汁": {"hydration_factor": 0.7, "notes": "含糖分，需适量"},
"运动饮料": {"hydration_factor": 1.1, "notes": "含电解质，适合大量出汗后"}
}
def evaluate_hydration_status(self, user_data: Dict) -> List[Dict]:
"""评估饮水状态并返回建议"""
recommendations = []
for rule in self.rules:
try:
if rule["condition"](user_data):
recommendations.append({
"name": rule["name"],
"action": rule["action"],
"severity": rule["severity"],
"message": rule["message"],
"timestamp": datetime.datetime.now()
})
except Exception as e:
logger.warning(f"规则执行失败: {rule['name']}, 错误: {e}")
# 按严重程度排序
severity_order = {"high": 3, "medium": 2, "low": 1}
recommendations.sort(key=lambda x: severity_order[x["severity"]], reverse=True)
return recommendations[:5] # 返回前5条建议
def get_water_source_advice(self, water_type: WaterSource, amount: float) -> Dict:
"""获取不同水源的饮水建议"""
if water_type.value in self.water_source_effects:
effect = self.water_source_effects[water_type.value]
return {
"type": water_type.value,
"effective_hydration": amount * effect["hydration_factor"],
"notes": effect["notes"],
"recommendation": f"饮水{amount}ml，实际补水效果{amount * effect['hydration_factor']:.0f}ml"
}
return {"type": water_type.value, "effective_hydration": amount, "notes": "无特别建议"}
class AdaptiveScheduler:
"""
自适应饮水计划调度器
基于强化学习优化饮水时间
"""
def __init__(self, user_profile: UserProfile):
self.user = user_profile
self.learning_rate = 0.1
self.discount_factor = 0.9
self.q_table = {} # Q-learning表
self.optimal_times = self._initialize_schedule()
def _initialize_schedule(self) -> List[Dict]:
"""初始化饮水时间表"""
# 基于用户作息时间生成初始时间表
wake_up = self.user.wake_up_time
sleep_time = self.user.sleep_time
# 将清醒时间分为8个时段
wake_hour = wake_up.hour
sleep_hour = sleep_time.hour if sleep_time.hour > wake_hour else sleep_time.hour + 24
total_hours = sleep_hour - wake_hour
interval = total_hours / 8
schedule = []
for i in range(8):
hour = wake_hour + interval * (i + 0.5) # 时段中间
if hour >= 24:
hour -= 24
schedule.append({
"hour": hour,
"amount": 200, # 初始建议每次200ml
"priority": 1.0,
"last_compliance": 0, # 上次遵守情况
"success_count": 0,
"total_count": 0
})
return schedule
def update_schedule(self, actual_intake: List[WaterIntake],
compliance_data: Dict) -> List[Dict]:
"""根据实际饮水情况更新时间表"""
if not actual_intake:
return self.optimal_times
# 分析实际饮水时间分布
hourly_intake = {}
for intake in actual_intake:
hour = intake.timestamp.hour
hourly_intake[hour] = hourly_intake.get(hour, 0) + intake.amount
# 更新Q表和学习
for i, slot in enumerate(self.optimal_times):
hour = int(slot["hour"])
actual_amount = hourly_intake.get(hour, 0)
# 计算奖励
if actual_amount >= slot["amount"] * 0.8: # 达到80%算成功
reward = 1
slot["success_count"] += 1
else:
reward = -0.5
slot["total_count"] += 1
success_rate = slot["success_count"] / max(slot["total_count"], 1)
# 更新优先级（Q-learning简化版）
old_priority = slot["priority"]
if hour in self.q_table:
old_q = self.q_table[hour]
new_q = old_q + self.learning_rate * (reward + self.discount_factor * success_rate - old_q)
self.q_table[hour] = new_q
else:
self.q_table[hour] = reward
# 根据成功率调整
if slot["total_count"] > 5: # 有足够数据后调整
if success_rate < 0.3: # 成功率低
# 调整到相邻时间段
self.optimal_times[i]["hour"] = (hour + 1) % 24
elif success_rate > 0.8: # 成功率高
# 稍微增加饮水量
self.optimal_times[i]["amount"] = min(300, slot["amount"] + 20)
slot["priority"] = success_rate
# 按优先级排序
self.optimal_times.sort(key=lambda x: x["priority"], reverse=True)
return self.optimal_times
def get_next_reminder(self, current_time: datetime.datetime) -> Optional[Dict]:
"""获取下一个提醒时间"""
current_hour = current_time.hour + current_time.minute / 60
for slot in self.optimal_times:
if slot["hour"] > current_hour:
# 转换为具体时间
reminder_time = current_time.replace(
hour=int(slot["hour"]),
minute=int((slot["hour"] % 1) * 60),
second=0
)
# 如果时间已过，推到明天
if reminder_time < current_time:
reminder_time += datetime.timedelta(days=1)
return {
"time": reminder_time,
"amount": slot["amount"],
"message": f"该喝水啦！建议饮用{slot['amount']}ml"
}
return None
class HydrationMonitor:
"""
饮水监测器
实时监测饮水状态
"""
def __init__(self, user_profile: UserProfile):
self.user = user_profile
self.daily_goal = self._calculate_daily_goal()
self.current_intake = 0
self.intake_history = []
self.sweat_history = []
self.environment_history = []
self.start_time = datetime.datetime.now().replace(hour=0, minute=0, second=0)
def _calculate_daily_goal(self) -> float:
"""计算每日饮水目标"""
# 基础需水量
base_need = HydrationCalculator
如果你觉得这个工具好用，欢迎关注我！
