远程患者监护方案
方案概述
远程患者监护（RPM，Remote Patient Monitoring）通过 IoT 设备持续采集患者生命体征，实现院外慢病管理和术后康复监护。
监护参数与设备
| 参数 | 设备 | 通信 | 采集频率 |
|---|---|---|---|
| 血压 | 电子血压计 | BLE / Cat.1 | 每天2次 |
| 血糖 | 连续血糖仪（CGM） | BLE | 每5分钟 |
| 心电 | 心电贴片 | BLE / Cat.1 | 连续 |
| 血氧 | 指夹式血氧仪 | BLE | 按需 |
| 体温 | 体温贴片 | BLE | 每小时 |
| 体重 | 智能体重秤 | BLE / Wi-Fi | 每天 |
| 活动量 | 手环/手表 | BLE | 连续 |
系统架构
患者端：
各类监测设备（BLE）
↓ BLE
手机 App（数据汇聚）
↓ Cat.1 / Wi-Fi
医疗云平台
或直连方案（无需手机）：
监测设备（Cat.1 直连）
↓ Cat.1 网络
医疗云平台
↓
医生工作站 / 护士站 / 家属 App慢病管理平台
高血压管理
python
class HypertensionManager:
"""高血压慢病管理"""
# 血压分级（中国高血压防治指南 2018）
BP_GRADES = [
(0, 120, 0, 80, "正常血压", "green"),
(120, 130, 80, 85, "正常高值", "yellow"),
(130, 140, 85, 90, "正常高值", "yellow"),
(140, 160, 90, 100, "1级高血压（轻度）", "orange"),
(160, 180, 100, 110, "2级高血压（中度）", "red"),
(180, 999, 110, 999, "3级高血压（重度）", "darkred"),
]
def classify_bp(self, systolic, diastolic):
"""血压分级"""
for sys_lo, sys_hi, dia_lo, dia_hi, grade, color in self.BP_GRADES:
if sys_lo <= systolic < sys_hi or dia_lo <= diastolic < dia_hi:
return {"grade": grade, "color": color}
return {"grade": "未知", "color": "gray"}
def analyze_trend(self, bp_records):
"""血压趋势分析（最近7天）"""
if len(bp_records) < 3:
return {"trend": "数据不足", "recommendation": "请继续监测"}
recent = bp_records[-7:]
sys_values = [r['systolic'] for r in recent]
dia_values = [r['diastolic'] for r in recent]
# 线性回归判断趋势
import numpy as np
x = np.arange(len(sys_values))
sys_slope = np.polyfit(x, sys_values, 1)[0]
if sys_slope > 2:
trend = "上升"
recommendation = "血压持续升高，建议就医调整用药"
elif sys_slope < -2:
trend = "下降"
recommendation = "血压控制良好，继续保持"
else:
trend = "平稳"
recommendation = "血压稳定，继续按时服药监测"
avg_sys = np.mean(sys_values)
avg_dia = np.mean(dia_values)
return {
"trend": trend,
"avg_systolic": round(avg_sys, 1),
"avg_diastolic": round(avg_dia, 1),
"recommendation": recommendation,
"morning_hypertension": self._check_morning_hypertension(bp_records)
}
def _check_morning_hypertension(self, records):
"""检测晨峰高血压"""
morning_records = [r for r in records
if 6 <= r['hour'] <= 10]
night_records = [r for r in records
if 0 <= r['hour'] <= 4]
if morning_records and night_records:
morning_avg = sum(r['systolic'] for r in morning_records) / len(morning_records)
night_avg = sum(r['systolic'] for r in night_records) / len(night_records)
if morning_avg - night_avg > 35:
return True
return False
def generate_report(self, patient_id, bp_records, period_days=30):
"""生成月度血压报告"""
recent = [r for r in bp_records
if r['timestamp'] > time.time() - period_days * 86400]
if not recent:
return None
sys_values = [r['systolic'] for r in recent]
dia_values = [r['diastolic'] for r in recent]
# 达标率（收缩压 < 140 且 舒张压 < 90）
controlled = sum(1 for r in recent
if r['systolic'] < 140 and r['diastolic'] < 90)
control_rate = controlled / len(recent) * 100
return {
"patient_id": patient_id,
"period": f"最近{period_days}天",
"measurement_count": len(recent),
"avg_systolic": round(sum(sys_values) / len(sys_values), 1),
"avg_diastolic": round(sum(dia_values) / len(dia_values), 1),
"max_systolic": max(sys_values),
"min_systolic": min(sys_values),
"control_rate": round(control_rate, 1),
"trend": self.analyze_trend(recent)
}糖尿病管理（CGM 连续血糖）
python
class DiabetesManager:
"""糖尿病连续血糖管理"""
# 血糖目标范围（mmol/L）
TARGET_RANGE = (3.9, 10.0) # 目标范围
LOW_THRESHOLD = 3.9 # 低血糖阈值
HIGH_THRESHOLD = 10.0 # 高血糖阈值
CRITICAL_LOW = 3.0 # 严重低血糖
CRITICAL_HIGH = 16.7 # 严重高血糖
def analyze_cgm(self, glucose_records):
"""分析 CGM 数据"""
values = [r['glucose'] for r in glucose_records]
# TIR（目标范围内时间）
in_range = sum(1 for v in values
if self.TARGET_RANGE[0] <= v <= self.TARGET_RANGE[1])
tir = in_range / len(values) * 100
# TBR（低于目标范围时间）
below_range = sum(1 for v in values if v < self.TARGET_RANGE[0])
tbr = below_range / len(values) * 100
# TAR（高于目标范围时间）
above_range = sum(1 for v in values if v > self.TARGET_RANGE[1])
tar = above_range / len(values) * 100
# GMI（葡萄糖管理指标，估算 HbA1c）
avg_glucose = sum(values) / len(values)
gmi = 3.31 + 0.02392 * avg_glucose * 18 # 转换为 mg/dL 计算
# CV（变异系数）
import statistics
cv = statistics.stdev(values) / avg_glucose * 100
return {
"avg_glucose_mmol": round(avg_glucose, 1),
"gmi_percent": round(gmi, 1),
"tir_percent": round(tir, 1),
"tbr_percent": round(tbr, 1),
"tar_percent": round(tar, 1),
"cv_percent": round(cv, 1),
"glucose_variability": "低" if cv < 36 else "高"
}
def detect_hypoglycemia(self, current_glucose, trend_arrow):
"""低血糖预警"""
alerts = []
if current_glucose < self.CRITICAL_LOW:
alerts.append({
"level": "critical",
"message": f"严重低血糖！当前血糖 {current_glucose} mmol/L，请立即补充糖分并就医"
})
elif current_glucose < self.LOW_THRESHOLD:
alerts.append({
"level": "warning",
"message": f"低血糖！当前血糖 {current_glucose} mmol/L，请补充糖分"
})
elif current_glucose < 4.5 and trend_arrow in ["↓↓", "↓"]:
# 血糖下降趋势，预测性低血糖预警
alerts.append({
"level": "info",
"message": f"血糖下降趋势，当前 {current_glucose} mmol/L，注意预防低血糖"
})
return alerts数据安全与隐私
python
import hashlib
import hmac
from cryptography.fernet import Fernet
class MedicalDataSecurity:
"""医疗数据安全处理"""
def __init__(self, encryption_key):
self.fernet = Fernet(encryption_key)
def encrypt_patient_data(self, data):
"""加密患者数据"""
import json
plaintext = json.dumps(data).encode()
return self.fernet.encrypt(plaintext)
def decrypt_patient_data(self, encrypted_data):
"""解密患者数据"""
import json
plaintext = self.fernet.decrypt(encrypted_data)
return json.loads(plaintext)
def anonymize(self, patient_data):
"""数据脱敏（用于研究分析）"""
anonymized = patient_data.copy()
# 替换姓名
anonymized['name'] = "患者" + hashlib.md5(
patient_data['name'].encode()
).hexdigest()[:6]
# 模糊化出生日期（只保留年份）
if 'birth_date' in anonymized:
anonymized['birth_date'] = anonymized['birth_date'][:4] + "-01-01"
# 删除联系方式
for field in ['phone', 'email', 'address', 'id_card']:
anonymized.pop(field, None)
return anonymized
def audit_log(self, user_id, action, patient_id, data_type):
"""记录数据访问审计日志"""
import time
log = {
"timestamp": time.time(),
"user_id": user_id,
"action": action, # read/write/export/delete
"patient_id": patient_id,
"data_type": data_type,
"ip_address": get_client_ip()
}
# 写入不可篡改的审计日志
save_audit_log(log)