352 lines
10 KiB
Python
352 lines
10 KiB
Python
# -*- 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)
|
|
|