物联网实践与探索 医疗设施新功能——基于行空板K10的病床输液监护系统 中等

class AppConfig:
    """应用程序全局配置"""
    
    # WiFi配置
    WIFI_SSID = "GOCLOUD"          # 无线网络名称
    WIFI_PASSWORD = "L2100Hbb"     # WiFi密码
    WIFI_TIMEOUT_MS = 50000        # 连接超时（毫秒）
    WIFI_MAX_RETRIES = 3           # 最大重试次数
    
    # MQTT配置
    MQTT_SERVER = "192.168.123.16"       # 消息代理地址
    MQTT_PORT = 1883                     # 服务端口
    MQTT_USER = "siot"                   # 认证用户名
    MQTT_PASSWORD = "dfrobot"            # 认证密码
    MQTT_MAX_RETRIES = 3                 # 连接重试次数
    ALARM_TOPIC = "siot/infusion_alarm"  # 输液报警主题
    START_TOPIC = "siot/infusion_start"  # 输液开始主题
    END_TOPIC = "siot/infusion_end"      # 输液结束主题
    
    # 硬件参数
    HARDWARE_DEBOUNCE_MS = 50      # 传感器消抖时间（毫秒）
    DISPLAY_REFRESH_MS = 1000      # 屏幕刷新间隔（毫秒）
    BUZZER_FREQ = 2000             # 蜂鸣器频率（Hz）
    BUZZER_DURATION_MS = 500       # 蜂鸣持续时间（毫秒）
    SENSOR_PIN = 0                 # 滴液传感器GPIO引脚
    SPEAKER_PIN = 1                # 蜂鸣器控制引脚
    
    # 安全参数（成人标准）
    SAFETY_MIN_DROP_RATE = 40      # 最低安全滴速（滴/分钟）
    SAFETY_MAX_DROP_RATE = 60      # 最高安全滴速（滴/分钟）

class InfusionMetrics:
    """输液过程指标数据容器
    
    属性：
    - drop_rate: 当前滴速（滴/分钟）
    - elapsed_time: 已输注时间（HH:MM:SS）
    - remaining_volume: 剩余液量（毫升）
    - remaining_time: 预计剩余时间（HH:MM:SS）
    """
    def __init__(self, drop_rate, elapsed_time, remaining_volume, remaining_time):
        self.drop_rate = drop_rate
        self.elapsed_time = elapsed_time
        self.remaining_volume = remaining_volume
        self.remaining_time = remaining_time

class InfusionCalculator:
    """输液指标计算引擎"""
    
    @staticmethod
    def calculate(current_time, start_time, drop_count, total_ml, drop_factor, cfg):
        """执行所有指标计算
        
        参数：
        current_time -- 当前时间（毫秒ticks）
        start_time -- 输液开始时间（毫秒ticks）
        drop_count -- 累计滴数
        total_ml -- 输液总量（毫升）
        drop_factor -- 滴系数（滴/毫升）
        cfg -- 配置对象
        
        返回：
        InfusionMetrics对象包含计算结果
        """
        # 计算经过时间（防止除零）
        elapsed_ticks = max(time.ticks_diff(current_time, start_time), 1)
        elapsed_min = elapsed_ticks / 60000  # 转换为分钟
        
        # 有效滴数处理（防止负数）
        valid_drops = max(drop_count, 0)
        
        # 计算原始滴速并约束到安全范围
        raw_rate = valid_drops / elapsed_min
        safe_rate = clamp(raw_rate, cfg.SAFETY_MIN_DROP_RATE, cfg.SAFETY_MAX_DROP_RATE)
        
        # 计算液量
        infused_ml = min(valid_drops / drop_factor, total_ml)
        remaining_ml = max(total_ml - infused_ml, 0)
        
        # 估算剩余时间（基于安全滴速）
        remain_min = (remaining_ml * drop_factor / safe_rate) if safe_rate > 0 else 0
        
        return InfusionMetrics(
            safe_rate,
            format_duration(elapsed_min),
            remaining_ml,
            format_duration(remain_min)
        )

def _handle_sensor(self, current_time):
    """处理滴液传感器信号（带消抖）"""
    current_state = self.drop_sensor.read_digital()
    
    # 状态变化检测
    if current_state != self.last_sensor_state:
        # 消抖时间检查
        if time.ticks_diff(current_time, self.last_debounce_time) > self.cfg.HARDWARE_DEBOUNCE_MS:
            # 下降沿触发计数（传感器输出低电平）
            if current_state == 0:
                max_drops = self.total_volume * self.drop_factor
                if self.drop_count < max_drops:
                    self.drop_count += 1
                    print(f"[传感器] 滴数更新：{self.drop_count}")
            
            # 更新状态记录
            self.last_sensor_state = current_state
            self.last_debounce_time = current_time

def _render_metrics(self, metrics):
    """绘制动态监测指标"""
    y_pos = [90, 140, 190, 240]  # 四行数据的Y坐标
    self.display.print("Rate: {:.1f}/min".format(metrics.drop_rate), 20, y_pos[0])
    self.display.print("Elapsed: {}".format(metrics.elapsed_time), 20, y_pos[1])
    self.display.print("Volume: {:.1f}mL".format(metrics.remaining_volume), 20, y_pos[2])
    self.display.print("Remain: {}".format(metrics.remaining_time), 20, y_pos[3])

def _handle_sensor(self, current_time):
    """模拟滴液生成（替代真实传感器处理）"""
    if current_time >= self.next_drop_time:
        # 生成随机间隔（±20%波动）
        variance = random.uniform(0.8, 1.2)
        interval = (60000 // self.base_rate) * variance
        self.next_drop_time = current_time + int(interval)
        
        # 安全增加滴数
        max_drops = self.total_volume * self.drop_factor
        if self.drop_count < max_drops:
            self.drop_count += 1
            print(f"[模拟] 滴数：{self.drop_count}")

def _update_interface(self, current_time):
    """增强界面更新（添加控制台输出）"""
    metrics = super()._update_interface(current_time)
    
    # 控制台打印监控数据
    print("\n==== 实时监控 =====")
    print(f"当前滴速: {metrics.drop_rate:.1f} 滴/分钟")
    print(f"已用时间: {metrics.elapsed_time}")
    print(f"剩余液量: {metrics.remaining_volume:.1f}mL")
    print(f"预计剩余: {metrics.remaining_time}")
    print("===================\n")      
    return metrics

def _process_alarm(self, msg):
    """处理报警消息
    消息格式："Bed-1|low volume:0.95mL"
    """
    try:
        # 先按 | 分割消息
        parts = msg.split('|')
        # 提取床号部分，例如从 Bed-1 中提取 1
        bed_name = parts[0]
        alarm_msg = parts[1]
        bed_number = int(bed_name.split('-')[1])
        print(f"[报警] 低液量：{bed_name} {alarm_msg}")
        self.alarm_beds.add(bed_name)
        self.last_alarm_time = time.ticks_ms()
        self.alarm_flashing = True  # 开启闪烁标志
        # 文字闪烁逻辑
        for _ in range(5):  # 闪烁5次，可根据需要调整
            self._trigger_buzzer()
            self.display.print("Low Alarm", self.cfg.x_coordinates[bed_number - 1], self.cfg.y_coordinates[bed_number - 1])
            time.sleep(0.5)
            self.display.print("         ", self.cfg.x_coordinates[bed_number - 1], self.cfg.y_coordinates[bed_number - 1])
            time.sleep(0.5)
        self.alarm_flashing = False  # 关闭闪烁标志
    except ValueError as e:
        print(f"消息解析错误，无法将床号转换为整数: {e}")
    except IndexError as e:
        print(f"消息格式错误，无法正确分割消息: {e}")
    except Exception as e:
        print(f"消息处理异常: {e}")

def _process_start(self, msg):
    """处理报警消息
    消息格式："Bed-1|start time:2025-02-06 12:28:15"
    """
    try:
        bed_name, start_time = msg.split('|')
        print(f"[事件] {bed_name} {start_time}")
        bed_number = int(bed_name.split('-')[1])  # 分割获取床号
        self.display.print(bed_name, self.cfg.x_coordinates[bed_number - 1]+10, self.cfg.y_coordinates[bed_number - 1] - 50)
    except Exception as e:
        print("消息解析失败:", e)

def _process_end(self, msg):
    """处理报警消息
    消息格式："Bed-1|end time:2025-02-06 12:28:15"
    """
    try:
        bed_name, end_time = msg.split('|')
        print(f"[事件] {bed_name} {end_time}")
        bed_number = int(bed_name.split('-')[1])  # 分割获取床号
        self.display.print("      ", self.cfg.x_coordinates[bed_number - 1]+10, self.cfg.y_coordinates[bed_number - 1] - 50)
    except Exception as e:
        print("消息解析失败:", e)

def _on_alarm(self, topic, msg):
    """报警消息回调"""
    try:
        bed_name, alarm_msg = msg.split('|')
        print(f"[网络] {topic}：{bed_name} {alarm_msg}")

        # 更新报警状态
        self.alarm_beds.add(bed_name)
        self.last_alarm_time = time.ticks_ms()

        # 触发声音报警
        self._trigger_buzzer()
    except:
        print("无效报警消息:", msg)

def _display_all_beds(self):
    """显示所有床位"""
    self._draw_multiple_beds(self.cfg.BED_COLUMNS, self.cfg.BED_ROWS, self.cfg.RECT_WIDTH, self.cfg.RECT_HEIGHT)