diff --git a/onboardprogram/slave.py b/onboardprogram/slave.py new file mode 100644 index 0000000..f9976ae --- /dev/null +++ b/onboardprogram/slave.py @@ -0,0 +1,351 @@ +# -*- coding: utf-8 -*- +# 智能生活心率/摔倒监测从机(第二块 mPython 掌控板) +# 主从机都连接 Wi-Fi,通过 UDP 通信;不接 UART 线。 +# 心率是模拟数据,不是PPG/ECG真实测量值;摔倒判定为演示级逻辑。 + +from mpython import * +import socket +import time +import json +import random +import math + +# -------------------- Wi-Fi配置 -------------------- +# 与主机 main.py 填写相同的路由器 Wi-Fi 账号。 +WIFI_SSID = "48-502" +WIFI_PASSWORD = "a13356041516" + +# 主机连接路由器失败时会创建此备用AP;从机会自动尝试加入。 +AP_SSID = "QHZXManagement" +AP_PASSWORD = "QHZX2026" +WIFI_CONNECT_TIMEOUT = 6 +RECONNECT_INTERVAL_MS = 10000 + +# -------------------- UDP配置 -------------------- +MASTER_UDP_PORT = 4210 # 主机接收从机状态广播的端口 +SLAVE_UDP_PORT = 4211 # 从机绑定端口,同时接收主机控制命令 +SEND_INTERVAL_MS = 500 +MASTER_IP_OVERRIDE = "" # 如路由器屏蔽广播,可填写主机IP,例如 192.168.1.20 + +# -------------------- 模拟心率配置 -------------------- +HEART_RATE_MIN = 62 +HEART_RATE_MAX = 96 +HEART_RATE_UPDATE_MS = 1800 + +# -------------------- 摔倒演示检测参数 -------------------- +ACCEL_SAMPLE_MS = 50 +FALL_IMPACT_DELTA_G = 1.35 +FALL_TILT_DEGREES = 62 +FALL_CONFIRM_MS = 900 +FALL_COOLDOWN_MS = 8000 + +heart_rate_bpm = 74 +fall_alarm = False +fall_count = 0 +fall_seq = 0 +last_hr_update = time.ticks_ms() +last_send = 0 +last_accel_sample = 0 +last_fall_ms = None +last_accel_magnitude = None +impact_candidate_ms = None +last_button_pressed = False + +my_wifi = wifi() +network_ok = False +network_mode = "OFFLINE" +local_ip = "0.0.0.0" +broadcast_ip = "255.255.255.255" +udp_sock = None +last_reconnect_ms = 0 +last_network_error = "" + + +def is_wifi_connected(): + try: + return bool(my_wifi.sta.isconnected()) + except Exception: + return False + + +def calc_broadcast(ifconfig): + """根据本机IP和子网掩码计算定向广播地址。""" + try: + ip_parts = [int(x) for x in ifconfig[0].split(".")] + mask_parts = [int(x) for x in ifconfig[1].split(".")] + result = [] + for ip_part, mask_part in zip(ip_parts, mask_parts): + result.append(str(ip_part | (255 - mask_part))) + return ".".join(result) + except Exception: + return "255.255.255.255" + + +def try_wifi(ssid, password): + global last_network_error + try: + my_wifi.connectWiFi(ssid, password, timeout=WIFI_CONNECT_TIMEOUT) + except Exception as e: + last_network_error = str(e) + print("Wi-Fi连接异常", ssid, e) + return is_wifi_connected() + + +def close_udp(): + global udp_sock + if udp_sock is not None: + try: + udp_sock.close() + except Exception: + pass + udp_sock = None + + +def setup_udp(): + global udp_sock, local_ip, broadcast_ip + try: + info = my_wifi.sta.ifconfig() + local_ip = info[0] + broadcast_ip = calc_broadcast(info) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + except Exception: + pass + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + except Exception: + # 部分 MicroPython 固件不提供 SO_BROADCAST 常量;仍尝试 sendto。 + pass + sock.bind(("0.0.0.0", SLAVE_UDP_PORT)) + sock.setblocking(False) + udp_sock = sock + print("从机UDP已启动,本机IP:", local_ip) + print("主机UDP广播地址:", broadcast_ip, MASTER_UDP_PORT) + return True + except Exception as e: + print("从机UDP初始化失败:", e) + close_udp() + return False + + +def connect_network(): + """先连家庭路由器;失败后连主机启动的 QHZXManagement AP。""" + global network_ok, network_mode, local_ip, broadcast_ip + + network_ok = False + close_udp() + + oled.fill(0) + oled.DispChar("心率监测从机", 0, 0) + oled.DispChar("连接路由WiFi...", 0, 16) + oled.show() + + if try_wifi(WIFI_SSID, WIFI_PASSWORD): + network_mode = "STA" + else: + oled.fill(0) + oled.DispChar("路由WiFi不可用", 0, 0) + oled.DispChar("连接主机管理AP", 0, 16) + oled.DispChar(AP_SSID, 0, 32) + oled.show() + + # 主机可能刚开始启动AP,分几次尝试加入。 + joined = False + for attempt in range(3): + if try_wifi(AP_SSID, AP_PASSWORD): + joined = True + break + time.sleep_ms(1500) + + if joined: + network_mode = "AP_CLIENT" + else: + network_mode = "OFFLINE" + network_ok = False + print("家庭Wi-Fi和主机备用AP均连接失败;稍后重试") + return False + + try: + local_ip = my_wifi.sta.ifconfig()[0] + broadcast_ip = calc_broadcast(my_wifi.sta.ifconfig()) + except Exception: + local_ip = "0.0.0.0" + broadcast_ip = "255.255.255.255" + + if not setup_udp(): + network_ok = False + return False + + network_ok = True + print("从机Wi-Fi模式:", network_mode, "IP:", local_ip) + return True + + +def send_status(): + if not network_ok or udp_sock is None: + return + + packet = { + "type": "slave_status", + "transport": "UDP/Wi-Fi", + "heart_rate_bpm": int(heart_rate_bpm), + "heart_rate_simulated": True, + "fall_alarm": bool(fall_alarm), + "fall_count": int(fall_count), + "fall_seq": int(fall_seq), + "fall_source": "slave_accelerometer_demo", + "wifi_mode": network_mode, + "ip_address": local_ip, + "uptime": time.ticks_ms() // 1000, + } + + try: + payload = json.dumps(packet).encode("utf-8") + if MASTER_IP_OVERRIDE: + destination_ip = MASTER_IP_OVERRIDE + else: + destination_ip = broadcast_ip + udp_sock.sendto(payload, (destination_ip, MASTER_UDP_PORT)) + except Exception as e: + print("UDP发送状态失败:", e) + + +def poll_master_commands(): + global fall_alarm, impact_candidate_ms + if udp_sock is None: + return + + for _ in range(4): + try: + data, addr = udp_sock.recvfrom(512) + except OSError: + break + except Exception as e: + print("UDP接收异常:", e) + break + + try: + command = json.loads(data.decode("utf-8", "ignore")) + if command.get("cmd") == "reset_fall": + fall_alarm = False + impact_candidate_ms = None + print("收到主机UDP指令:清除当前摔倒报警") + send_status() + except Exception as e: + print("主机UDP命令解析异常:", e, "来源:", addr) + + +def simulate_heart_rate(now): + global heart_rate_bpm, last_hr_update + if time.ticks_diff(now, last_hr_update) >= HEART_RATE_UPDATE_MS: + last_hr_update = now + # 缓慢随机变化,只用于界面联调和演示。 + heart_rate_bpm += random.randint(-3, 3) + heart_rate_bpm = max(HEART_RATE_MIN, min(HEART_RATE_MAX, heart_rate_bpm)) + + +def trigger_fall(source="accelerometer_demo"): + global fall_alarm, fall_count, fall_seq, last_fall_ms, impact_candidate_ms + now = time.ticks_ms() + if last_fall_ms is not None and time.ticks_diff(now, last_fall_ms) < FALL_COOLDOWN_MS: + return + + fall_alarm = True + fall_count += 1 + fall_seq += 1 + last_fall_ms = now + impact_candidate_ms = None + print("疑似摔倒报警,从机来源:", source, "累计:", fall_count) + send_status() + + +def check_fall(now): + global last_accel_magnitude, impact_candidate_ms + try: + x = accelerometer.get_x() + y = accelerometer.get_y() + z = accelerometer.get_z() + magnitude = math.sqrt(x * x + y * y + z * z) + + if last_accel_magnitude is not None: + delta = abs(magnitude - last_accel_magnitude) + if delta >= FALL_IMPACT_DELTA_G: + impact_candidate_ms = now + + last_accel_magnitude = magnitude + + if impact_candidate_ms is not None: + elapsed = time.ticks_diff(now, impact_candidate_ms) + if elapsed >= FALL_CONFIRM_MS: + try: + roll, pitch = accelerometer.roll_pitch_angle() + tilted = ( + abs(roll) >= FALL_TILT_DEGREES + or abs(pitch) >= FALL_TILT_DEGREES + ) + except Exception: + tilted = False + + if tilted: + trigger_fall("accelerometer_impact_and_tilt") + elif elapsed >= 2000: + impact_candidate_ms = None + except Exception as e: + print("加速度计读取异常:", e) + + +def test_button(): + global last_button_pressed + try: + pressed = button_a.value() == 0 + except Exception: + pressed = False + + if pressed and not last_button_pressed: + trigger_fall("button_a_test") + last_button_pressed = pressed + + +# -------------------- 启动与主循环 -------------------- +connect_network() +last_reconnect_ms = time.ticks_ms() + +while True: + now = time.ticks_ms() + + # Wi-Fi中断时重新尝试:家庭路由器优先,其次主机备用AP。 + if not is_wifi_connected(): + network_ok = False + close_udp() + + if not network_ok and time.ticks_diff(now, last_reconnect_ms) >= RECONNECT_INTERVAL_MS: + last_reconnect_ms = now + connect_network() + + poll_master_commands() + simulate_heart_rate(now) + + if time.ticks_diff(now, last_accel_sample) >= ACCEL_SAMPLE_MS: + last_accel_sample = now + check_fall(now) + + test_button() + + if time.ticks_diff(now, last_send) >= SEND_INTERVAL_MS: + last_send = now + send_status() + + try: + oled.fill(0) + oled.DispChar("心率从机 UDP", 0, 0) + oled.DispChar("HR:%d BPM 模拟" % heart_rate_bpm, 0, 16) + oled.DispChar("摔倒报警!" if fall_alarm else "状态正常", 0, 32) + oled.DispChar("STA:%s" % local_ip if network_ok else "WiFi重连中", 0, 48) + oled.show() + except Exception: + pass + + time.sleep_ms(20) +