add: 上传掌控板从机程序源码slave.py

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)
+