Compare commits

...
11 Commits
28 changed files with 3984 additions and 2 deletions

No files matched your search

+18 -2
View File
@@ -1,3 +1,19 @@
# Competition-2026 # 智联安居——万物智联居家适老看护系统
智联安居——万物智联居家适老看护系统 ### 更多内容请查看[百科](https://qhzx-2026-competition-repo.20130604.xyz/Jiaxing-Qinghe-Middle-School/Competition-2026/wiki/Home)
## 作品说明表
> 民生服务线·助老助残创客作品
| 项目 | 内容 |
| ---- | ---- |
| 作品名称 | 智联安居——居家老人智能助老监护系统 |
| 参赛类别 | 民生服务线(助老助残类) |
| 主控单元 | ESP32 开源主控板 |
| 作品简介 | 本作品面向独居老人居家安全场景,搭建一体化智能监护小屋模型。集成人体感应照明、摔倒震动检测、一键紧急呼叫、识别摔倒震动时,系统报警;过道人体红外感应自动点亮照明,环境不佳自动通风,降低老人夜间起夜磕碰风险,用低成本开源硬件实现居家安全防护与适老化辅助。 |
| 硬件清单 | ESP32主控、人体红外传感器、震动传感器、燃气MQ-2传感器、独立按键(紧急呼叫)、舵机×2、蜂鸣器、LED指示灯、OLED显示屏、电源模块、亚克力房屋模型框架、连杆结构 |
| 软件/代码 | MicroPython,自主编写联动逻辑;传感器采集、阈值判断、舵机控制、声光告警、屏幕状态显示 |
| 核心功能 | 1. 人体红外感应:人体靠近自动开启过道照明<br>2. 燃气监测:检测燃气泄漏,声光报警+舵机自动开窗通风<br>3. 震动摔倒检测:震动触发紧急告警<br>4. 一键呼救:老人按下按钮,蜂鸣器+LED告警提示<br>5. 助起支架:舵机连杆驱动,模拟床边辅助扶手弹出<br>6. OLED屏幕实时展示设备状态:正常/告警 |
| 创新亮点 | 1. 多风险融合监测:同时覆盖摔倒、燃气、夜间出行多个居家高危场景,不只是单一报警<br>2. 主动处置而非仅告警:燃气超标自动开窗、机械辅助支架,实现“感知+执行”闭环<br>3. 低成本开源方案,使用通用创客元器件,易复刻、可拓展(后续可增加远程消息推送)<br>4. 适配居家适老化需求,贴合民生助老主题 |
| 拓展方向 | 增加WiFi联网,告警信息推送至家属手机;接入语音模块,实现语音求助、语音播报;增加温湿度监测 |
| 知识运用 | 传感器信号采集、模数转换、舵机伺服控制、嵌入式逻辑编程、物联网感知联动、机械连杆结构设计 |
+6
View File
@@ -0,0 +1,6 @@
.gradle/
build/
local.properties
.idea/
*.iml
app/build/
+6
View File
@@ -0,0 +1,6 @@
FallGuard project status
Source project: created.
Local APK build: NOT run in this environment because Android SDK / Gradle are not installed and outbound DNS/network access is unavailable.
APK build workflow: .github/workflows/android.yml
Expected debug APK output after a successful build: app/build/outputs/apk/debug/app-debug.apk
+71
View File
@@ -0,0 +1,71 @@
plugins {
id("com.android.application")
id("org.jetbrains.kotlin.android")
id("org.jetbrains.kotlin.plugin.compose")
}
android {
namespace = "com.zhouyi.elife"
compileSdk = 35
defaultConfig {
applicationId = "com.zhouyi.elife"
minSdk = 24
targetSdk = 35
versionCode = 2
versionName = "1.1.0"
}
buildTypes {
debug {
// Keep the requested package name com.zhouyi.elife even for debug APK.
versionNameSuffix = "-debug"
}
release {
isMinifyEnabled = false
proguardFiles(
getDefaultProguardFile("proguard-android-optimize.txt"),
"proguard-rules.pro"
)
}
}
buildFeatures {
compose = true
}
compileOptions {
sourceCompatibility = JavaVersion.VERSION_17
targetCompatibility = JavaVersion.VERSION_17
}
kotlinOptions {
jvmTarget = "17"
}
packaging {
resources {
excludes += "/META-INF/{AL2.0,LGPL2.1}"
}
}
}
dependencies {
val composeBom = platform("androidx.compose:compose-bom:2024.12.01")
implementation(composeBom)
androidTestImplementation(composeBom)
implementation("androidx.core:core-ktx:1.15.0")
implementation("androidx.activity:activity-compose:1.10.0")
implementation("androidx.lifecycle:lifecycle-runtime-ktx:2.8.7")
implementation("androidx.compose.ui:ui")
implementation("androidx.compose.ui:ui-graphics")
implementation("androidx.compose.ui:ui-tooling-preview")
implementation("androidx.compose.material3:material3")
implementation("androidx.compose.material:material-icons-extended")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.9.0")
debugImplementation("androidx.compose.ui:ui-tooling")
debugImplementation("androidx.compose.ui:ui-test-manifest")
androidTestImplementation("androidx.test.ext:junit:1.2.1")
androidTestImplementation("androidx.test.espresso:espresso-core:3.6.1")
androidTestImplementation("androidx.compose.ui:ui-test-junit4")
}
+1
View File
@@ -0,0 +1 @@
# No custom ProGuard rules are required for the current app.
+23
View File
@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android">
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<application
android:allowBackup="true"
android:icon="@drawable/ic_launcher"
android:label="智能生活"
android:supportsRtl="true"
android:theme="@style/Theme.SmartLife"
android:usesCleartextTraffic="true">
<activity
android:name=".MainActivity"
android:exported="true"
android:screenOrientation="portrait">
<intent-filter>
<action android:name="android.intent.action.MAIN" />
<category android:name="android.intent.category.LAUNCHER" />
</intent-filter>
</activity>
</application>
</manifest>
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,8 @@
<vector xmlns:android="http://schemas.android.com/apk/res/android"
android:width="48dp" android:height="48dp"
android:viewportWidth="48" android:viewportHeight="48">
<path android:fillColor="#536DFE" android:pathData="M5,22L24,5 43,22 39,26 24,13 9,26z" />
<path android:fillColor="#536DFE" android:pathData="M10,23 L38,23 L38,42 L10,42 Z" />
<path android:fillColor="#FFFFFF" android:pathData="M20,29L28,29L28,42L20,42Z" />
<path android:fillColor="#B8C5FF" android:pathData="M14,28L18,28L18,32L14,32Z M30,28L34,28L34,32L30,32Z" />
</vector>
+11
View File
@@ -0,0 +1,11 @@
<?xml version="1.0" encoding="utf-8"?>
<resources>
<style name="Theme.SmartLife" parent="android:style/Theme.Material.Light.NoActionBar">
<item name="android:fontFamily">sans</item>
<item name="android:colorAccent">#536DFE</item>
<item name="android:statusBarColor">#F3F6FC</item>
<item name="android:navigationBarColor">#F3F6FC</item>
<item name="android:windowLightStatusBar">true</item>
<item name="android:windowActionModeOverlay">true</item>
</style>
</resources>
+5
View File
@@ -0,0 +1,5 @@
plugins {
id("com.android.application") version "8.7.3" apply false
id("org.jetbrains.kotlin.android") version "2.0.21" apply false
id("org.jetbrains.kotlin.plugin.compose") version "2.0.21" apply false
}
+4
View File
@@ -0,0 +1,4 @@
org.gradle.jvmargs=-Xmx2g -Dfile.encoding=UTF-8
android.useAndroidX=true
kotlin.code.style=official
org.gradle.parallel=true
+7
View File
@@ -0,0 +1,7 @@
distributionBase=GRADLE_USER_HOME
distributionPath=manual-wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-8.9-bin.zip
networkTimeout=10000
validateDistributionUrl=true
zipStoreBase=GRADLE_USER_HOME
zipStorePath=manual-wrapper/dists
Vendored
+24
View File
@@ -0,0 +1,24 @@
#!/usr/bin/env bash
# Small wrapper bootstrap that downloads Gradle when it is not already cached.
set -euo pipefail
GRADLE_VERSION="8.9"
GRADLE_USER_HOME="${GRADLE_USER_HOME:-$HOME/.gradle}"
INSTALL_ROOT="$GRADLE_USER_HOME/manual-wrapper/gradle-$GRADLE_VERSION"
GRADLE_BIN="$INSTALL_ROOT/gradle-$GRADLE_VERSION/bin/gradle"
if [[ ! -x "$GRADLE_BIN" ]]; then
mkdir -p "$INSTALL_ROOT"
ARCHIVE="$INSTALL_ROOT/gradle-$GRADLE_VERSION-bin.zip"
URL="https://services.gradle.org/distributions/gradle-$GRADLE_VERSION-bin.zip"
echo "Gradle $GRADLE_VERSION 尚未安装,正在下载……"
if command -v curl >/dev/null 2>&1; then
curl --fail --location --retry 3 "$URL" --output "$ARCHIVE"
elif command -v wget >/dev/null 2>&1; then
wget -O "$ARCHIVE" "$URL"
else
echo "需要 curl 或 wget 才能下载 Gradle。" >&2
exit 1
fi
unzip -q -o "$ARCHIVE" -d "$INSTALL_ROOT"
rm -f "$ARCHIVE"
fi
exec "$GRADLE_BIN" "$@"
+12
View File
@@ -0,0 +1,12 @@
@echo off
setlocal
set "GRADLE_VERSION=8.9"
set "INSTALL_ROOT=%USERPROFILE%\.gradle\manual-wrapper\gradle-%GRADLE_VERSION%"
set "GRADLE_BIN=%INSTALL_ROOT%\gradle-%GRADLE_VERSION%\bin\gradle.bat"
if not exist "%GRADLE_BIN%" (
echo Gradle %GRADLE_VERSION% 尚未安装,正在下载……
powershell -NoProfile -ExecutionPolicy Bypass -Command "$ErrorActionPreference='Stop'; $root='%INSTALL_ROOT%'; New-Item -ItemType Directory -Force -Path $root | Out-Null; $zip=Join-Path $root 'gradle-%GRADLE_VERSION%-bin.zip'; Invoke-WebRequest -Uri 'https://services.gradle.org/distributions/gradle-%GRADLE_VERSION%-bin.zip' -OutFile $zip; Expand-Archive -Path $zip -DestinationPath $root -Force; Remove-Item $zip -Force"
if errorlevel 1 exit /b 1
)
call "%GRADLE_BIN%" %*
exit /b %ERRORLEVEL%
+18
View File
@@ -0,0 +1,18 @@
pluginManagement {
repositories {
google()
mavenCentral()
gradlePluginPortal()
}
}
dependencyResolutionManagement {
repositoriesMode.set(RepositoriesMode.FAIL_ON_PROJECT_REPOS)
repositories {
google()
mavenCentral()
}
}
rootProject.name = "SmartLife"
include(":app")
Binary file not shown.

Before

Width:  |  Height:  |  Size: 5.0 MiB

+709
View File
@@ -0,0 +1,709 @@
# -*- coding: utf-8 -*-
# 智能生活主机 main.py(mPython / 掌控板 MicroPython)
# 功能:P0人体感应、P7 RFID门灯、P13/P14两路照明联动、P5/P6风扇、
# SHT20温湿度、UDP心率/摔倒从机、摇晃报警、HTTP API、Wi-Fi失败后启动AP。
# 注意:心率由从机模拟;摇晃报警是演示级“疑似摔倒”判断,不是医学级跌倒检测。
# 电机/灯具必须通过驱动板、MOSFET或继电器供电,不能直接由GPIO带负载。
from mpython import *
from machine import Pin
from bluebit import SHT20, Scan_Rfid
import socket
import time
import json
# ==================== 配置 ====================
WIFI_SSID = "48-502"
WIFI_PASSWORD = "a13356041516"
AP_SSID = "QHZXManagement"
AP_PASSWORD = "QHZX2026"
WIFI_CONNECT_TIMEOUT = 8
PIR_PIN_NO = 0
DOOR_LIGHT_PIN_NO = 7
ROOM_LIGHT_PIN_NOS = (13, 14)
FAN_PIN_A_NO = 5
FAN_PIN_B_NO = 6
RFID_AUTHORIZED_UID = 1239985986
DEFAULT_FAN_TEMP = 28.0
DEFAULT_FAN_HUMIDITY = 75.0
DEFAULT_LIGHT_TIMEOUT = 30
DEFAULT_DOOR_TIMEOUT = 8
MASTER_UDP_PORT = 4210
SLAVE_UDP_PORT = 4211
UDP_STATUS_TIMEOUT_MS = 5000
# 摇晃检测:100ms采样。
# 连续两次明显变化,或一次强烈变化,触发报警。
SHAKE_DELTA_SQ = 0.65
SHAKE_STRONG_DELTA_SQ = 2.0
SHAKE_WINDOW_MS = 900
SHAKE_COOLDOWN_MS = 5000
fan_auto = True
pir_auto = True
fan_temp_threshold = DEFAULT_FAN_TEMP
fan_humidity_threshold = DEFAULT_FAN_HUMIDITY
light_timeout = DEFAULT_LIGHT_TIMEOUT
door_timeout = DEFAULT_DOOR_TIMEOUT
# ==================== 状态变量 ====================
fallen = False
fall_count = 0
fall_time = 0
fall_source = ""
last_accel = None
last_local_sample_ms = 0
last_local_fall_event_ms = None
shake_hit_count = 0
last_shake_hit_ms = None
temperature = None
humidity = None
sensor_ok = False
last_sensor_ms = 0
pir_now = False
last_motion_ms = time.ticks_ms()
door_light_deadline = None
rfid_last = "—"
rfid_authorized = False
rfid_card_present = False
rfid_available = True
heart_rate_bpm = None
heart_rate_simulated = True
slave_fall_alarm = False
slave_fall_count = 0
slave_connected = False
slave_last_rx_ms = None
last_slave_fall_alarm = False
slave_ip = ""
slave_peer_addr = None
slave_udp = None
wifi_mode = "STA"
ip = "0.0.0.0"
pin_warnings = []
# ==================== GPIO初始化 ====================
def add_warning(message):
if message not in pin_warnings:
pin_warnings.append(message)
def make_pin(pin_number, mode, pull=None):
try:
pin = MPythonPin(pin_number, mode, pull)
if mode == PinMode.OUT:
pin.write_digital(0)
return pin
except Exception as e:
message = "P%d不可用: %s" % (pin_number, str(e))
print(message)
add_warning(message)
return None
pir_pin = make_pin(PIR_PIN_NO, PinMode.IN, Pin.PULL_DOWN)
door_light_pin = make_pin(DOOR_LIGHT_PIN_NO, PinMode.OUT)
room_light_pins = [make_pin(n, PinMode.OUT) for n in ROOM_LIGHT_PIN_NOS]
fan_pin_a = make_pin(FAN_PIN_A_NO, PinMode.OUT)
fan_pin_b = make_pin(FAN_PIN_B_NO, PinMode.OUT)
fan_available = fan_pin_a is not None and fan_pin_b is not None
room_light_available = all(pin is not None for pin in room_light_pins)
outputs = {"door_light": False, "room_light": False, "fan": False}
try:
sht20 = SHT20()
except Exception as e:
sht20 = None
add_warning("SHT20初始化失败: %s" % str(e))
# ==================== 输出控制 ====================
def set_door_light(state):
if door_light_pin is None:
outputs["door_light"] = False
return False
try:
door_light_pin.write_digital(1 if state else 0)
outputs["door_light"] = bool(state)
return True
except Exception as e:
add_warning("P7门灯输出失败: %s" % str(e))
outputs["door_light"] = False
return False
def set_room_lights(state):
"""P13/P14作为同一组灯,调用一次就同时开或同时关。"""
results = []
for index, pin in enumerate(room_light_pins):
if pin is None:
results.append(False)
continue
try:
pin.write_digital(1 if state else 0)
results.append(True)
except Exception as e:
add_warning("P%d照明输出失败: %s" % (ROOM_LIGHT_PIN_NOS[index], str(e)))
results.append(False)
outputs["room_light"] = bool(state) and len(results) == 2 and all(results)
if not all(results):
add_warning("照明灯组存在不可用引脚,请检查P13/P14及板卡版本")
return len(results) == 2 and all(results)
def set_fan(state):
global fan_available
if not fan_available:
outputs["fan"] = False
return False
try:
if state:
fan_pin_a.write_digital(1)
fan_pin_b.write_digital(0)
else:
fan_pin_a.write_digital(0)
fan_pin_b.write_digital(0)
outputs["fan"] = bool(state)
return True
except Exception as e:
fan_available = False
outputs["fan"] = False
add_warning("P%d/P%d风扇控制失败: %s" % (FAN_PIN_A_NO, FAN_PIN_B_NO, str(e)))
return False
# ==================== 摔倒/摇晃报警 ====================
def register_fall(source, now=None):
"""锁存一次疑似摔倒报警,直到App调用/api/reset。"""
global fallen, fall_count, fall_time, fall_source
if fallen:
return False
if now is None:
now = time.ticks_ms()
fallen = True
fall_count += 1
fall_time = now // 1000
fall_source = source
print("摔倒报警:", source)
return True
def detect_shake(now):
"""根据加速度相邻采样变化判断摇晃;触发后保持fallen=True。"""
global last_accel, last_local_sample_ms
global last_local_fall_event_ms, shake_hit_count, last_shake_hit_ms
if time.ticks_diff(now, last_local_sample_ms) < 100:
return
last_local_sample_ms = now
try:
accel = (
accelerometer.get_x(),
accelerometer.get_y(),
accelerometer.get_z()
)
if last_accel is not None:
dx = accel[0] - last_accel[0]
dy = accel[1] - last_accel[1]
dz = accel[2] - last_accel[2]
change_sq = dx * dx + dy * dy + dz * dz
if change_sq >= SHAKE_DELTA_SQ:
if (last_shake_hit_ms is None or
time.ticks_diff(now, last_shake_hit_ms) > SHAKE_WINDOW_MS):
shake_hit_count = 1
else:
shake_hit_count += 1
last_shake_hit_ms = now
confirmed = (change_sq >= SHAKE_STRONG_DELTA_SQ or shake_hit_count >= 2)
cooldown_ok = (
last_local_fall_event_ms is None or
time.ticks_diff(now, last_local_fall_event_ms) >= SHAKE_COOLDOWN_MS
)
if confirmed and cooldown_ok and not fallen:
if register_fall("主机摇晃检测(疑似摔倒)", now):
last_local_fall_event_ms = now
shake_hit_count = 0
elif (last_shake_hit_ms is not None and
time.ticks_diff(now, last_shake_hit_ms) > SHAKE_WINDOW_MS):
shake_hit_count = 0
last_accel = accel
except Exception as e:
print("主机加速度计读取异常:", e)
# ==================== UDP主从通信 ====================
def udp_send_slave_command(command):
if slave_udp is None or slave_peer_addr is None:
return False
try:
payload = json.dumps(command).encode("utf-8")
for _ in range(3):
slave_udp.sendto(payload, slave_peer_addr)
time.sleep_ms(30)
return True
except Exception as e:
add_warning("UDP发送从机命令失败: %s" % str(e))
return False
def handle_slave_message(message, addr=None):
global heart_rate_bpm, heart_rate_simulated
global slave_fall_alarm, slave_fall_count, slave_connected
global slave_last_rx_ms, last_slave_fall_alarm
global slave_ip, slave_peer_addr
if not isinstance(message, dict) or message.get("type") != "slave_status":
return
now = time.ticks_ms()
slave_last_rx_ms = now
slave_connected = True
if addr is not None:
slave_ip = addr[0]
slave_peer_addr = (addr[0], SLAVE_UDP_PORT)
bpm = message.get("heart_rate_bpm")
if isinstance(bpm, (int, float)):
heart_rate_bpm = int(bpm)
heart_rate_simulated = bool(message.get("heart_rate_simulated", True))
try:
slave_fall_count = int(message.get("fall_count", slave_fall_count))
except Exception:
pass
new_alarm = bool(message.get("fall_alarm", False))
slave_fall_alarm = new_alarm
if new_alarm and not last_slave_fall_alarm:
register_fall("从机摇晃检测(疑似摔倒)", now)
last_slave_fall_alarm = new_alarm
def poll_slave_udp():
global slave_connected
if slave_udp is None:
slave_connected = False
return
for _ in range(6):
try:
packet, addr = slave_udp.recvfrom(1024)
except OSError:
break
except Exception as e:
add_warning("UDP接收异常: %s" % str(e))
break
try:
message = json.loads(packet.decode("utf-8", "ignore"))
handle_slave_message(message, addr)
except Exception as e:
print("从机UDP数据解析异常:", e)
if slave_last_rx_ms is None:
slave_connected = False
else:
slave_connected = time.ticks_diff(time.ticks_ms(), slave_last_rx_ms) < UDP_STATUS_TIMEOUT_MS
# ==================== HTTP API ====================
def parse_query(query_string):
result = {}
if not query_string:
return result
for item in query_string.split("&"):
if "=" in item:
key, value = item.split("=", 1)
result[key] = value.replace("+", " ")
elif item:
result[item] = ""
return result
def status_data():
return {
"ok": True,
"message": "ok",
"temperature": temperature,
"humidity": humidity,
"sensor_ok": bool(sensor_ok),
"pir": bool(pir_now),
"pir_auto": bool(pir_auto),
"door_light": bool(outputs["door_light"]),
"room_light": bool(outputs["room_light"]),
"fan": bool(outputs["fan"]),
"fan_auto": bool(fan_auto),
"fan_available": bool(fan_available),
"room_light_available": bool(room_light_available),
"fan_temp_threshold": fan_temp_threshold,
"fan_humidity_threshold": fan_humidity_threshold,
"light_timeout": light_timeout,
"door_timeout": door_timeout,
"rfid_last": str(rfid_last),
"rfid_authorized": bool(rfid_authorized),
"rfid_available": bool(rfid_available),
"heart_rate_bpm": heart_rate_bpm,
"heart_rate_simulated": bool(heart_rate_simulated),
"slave_connected": bool(slave_connected),
"slave_transport": "UDP/Wi-Fi",
"slave_ip": slave_ip,
"slave_udp_port": SLAVE_UDP_PORT,
"slave_fall_alarm": bool(slave_fall_alarm),
"slave_fall_count": slave_fall_count,
"fall_source": fall_source,
"fall_alert": bool(fallen),
"fall_status": "摔倒报警" if fallen else "正常",
"shake_detection": True,
"pin_status": ";".join(pin_warnings) if pin_warnings else "所有已配置引脚初始化正常",
"uptime": time.ticks_ms() // 1000,
"fallen": bool(fallen),
"fall_count": fall_count,
"fall_time": fall_time,
"wifi_mode": wifi_mode,
"ip_address": ip,
"ap_ssid": AP_SSID,
"ap_password": AP_PASSWORD if wifi_mode == "AP" else ""
}
def send_json(conn, data, status_code="200 OK"):
body = json.dumps(data).encode("utf-8")
header = (
"HTTP/1.1 %s\r\n" % status_code
+ "Content-Type: application/json; charset=utf-8\r\n"
+ "Access-Control-Allow-Origin: *\r\n"
+ "Access-Control-Allow-Methods: GET, OPTIONS\r\n"
+ "Cache-Control: no-store\r\n"
+ "Connection: close\r\n"
+ "Content-Length: %d\r\n\r\n" % len(body)
)
try:
conn.send(header.encode("utf-8") + body)
except Exception:
pass
def control_device(params):
global fan_auto, pir_auto, door_light_deadline, last_motion_ms
device = params.get("device", "")
state = params.get("state", "0") in ("1", "true", "on")
if device == "fan":
fan_auto = False
if not fan_available:
return False, "风扇驱动不可用,请检查P5/P6与驱动板"
if not set_fan(state):
return False, "风扇控制失败,请检查电机驱动模块"
return True, "风扇状态已更新"
if device == "room_light":
pir_auto = False
if not set_room_lights(state):
return False, "照明灯组未能全部切换,请检查P13/P14"
if state:
last_motion_ms = time.ticks_ms()
return True, "两路照明已同步打开" if state else "两路照明已同步关闭"
if device == "door_light":
door_light_deadline = None
if not set_door_light(state):
return False, "P7门灯输出不可用,请检查接线"
return True, "门上灯状态已更新"
return False, "未知设备名称"
def update_settings(params):
global pir_auto, fan_auto
global fan_temp_threshold, fan_humidity_threshold
global light_timeout, door_timeout
try:
if "pir_auto" in params:
pir_auto = params["pir_auto"] in ("1", "true", "on")
if "fan_auto" in params:
fan_auto = params["fan_auto"] in ("1", "true", "on")
if "fan_temp" in params:
value = float(params["fan_temp"])
if not 15 <= value <= 40:
return False, "温度阈值范围为15到40°C"
fan_temp_threshold = value
if "fan_humidity" in params:
value = float(params["fan_humidity"])
if not 30 <= value <= 95:
return False, "湿度阈值范围为30到95%"
fan_humidity_threshold = value
if "light_timeout" in params:
light_timeout = max(5, min(600, int(params["light_timeout"])))
if "door_timeout" in params:
door_timeout = max(2, min(120, int(params["door_timeout"])))
return True, "自动化规则已保存"
except Exception as e:
return False, "参数无效: %s" % str(e)
def handle_request(conn):
global fallen, fall_source
global shake_hit_count, last_shake_hit_ms, last_local_fall_event_ms
global last_slave_fall_alarm
try:
conn.settimeout(1)
raw = conn.recv(1200)
if not raw:
return
request_line = raw.decode("utf-8", "ignore").split("\r\n", 1)[0]
parts = request_line.split(" ")
if len(parts) < 2:
send_json(conn, {"ok": False, "message": "bad request"}, "400 Bad Request")
return
target = parts[1]
if "?" in target:
path, query = target.split("?", 1)
else:
path, query = target, ""
params = parse_query(query)
if path == "/api/status":
send_json(conn, status_data())
elif path == "/api/control":
ok, message = control_device(params)
data = status_data()
data["ok"] = ok
data["message"] = message
send_json(conn, data, "200 OK" if ok else "400 Bad Request")
elif path == "/api/settings":
ok, message = update_settings(params)
data = status_data()
data["ok"] = ok
data["message"] = message
send_json(conn, data, "200 OK" if ok else "400 Bad Request")
elif path == "/api/reset":
fallen = False
fall_source = ""
shake_hit_count = 0
last_shake_hit_ms = None
last_local_fall_event_ms = time.ticks_ms()
last_slave_fall_alarm = False
udp_send_slave_command({"cmd": "reset_fall"})
data = status_data()
data["message"] = "主机和从机的当前摔倒标记已清除;累计次数保留"
send_json(conn, data)
elif path == "/":
page = (
"<!doctype html><html><head><meta charset='utf-8'>"
"<meta name='viewport' content='width=device-width,initial-scale=1'>"
"<title>智能生活</title></head><body style='font-family:sans-serif;"
"max-width:640px;margin:40px auto;padding:0 16px'>"
"<h1>智能生活主机已启动</h1>"
"<p><a href='/api/status'>查看设备状态</a></p>"
"<p>Wi-Fi模式:%s</p><p>地址:%s</p>"
"</body></html>"
) % (wifi_mode, ip)
page_bytes = page.encode("utf-8")
header = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/html; charset=utf-8\r\n"
"Connection: close\r\n"
"Content-Length: %d\r\n\r\n" % len(page_bytes)
)
conn.send(header.encode("utf-8") + page_bytes)
elif path == "/favicon.ico" or parts[0] == "OPTIONS":
send_json(conn, {"ok": True})
else:
send_json(conn, {"ok": False, "message": "not found"}, "404 Not Found")
except Exception as e:
print("HTTP请求处理异常:", e)
finally:
try:
conn.close()
except Exception:
pass
# ==================== Wi-Fi连接:失败后启动备用AP ====================
oled.fill(0)
oled.DispChar("智能生活主机", 0, 0)
oled.DispChar("连接路由WiFi...", 0, 16)
oled.show()
my_wifi = wifi()
try:
my_wifi.connectWiFi(WIFI_SSID, WIFI_PASSWORD, timeout=WIFI_CONNECT_TIMEOUT)
except Exception as e:
print("STA连接异常:", e)
sta_ok = False
try:
sta_ok = my_wifi.sta.isconnected()
except Exception:
sta_ok = False
if sta_ok:
wifi_mode = "STA"
try:
ip = my_wifi.sta.ifconfig()[0]
except Exception:
ip = "0.0.0.0"
else:
wifi_mode = "AP"
oled.fill(0)
oled.DispChar("路由WiFi不可用", 0, 0)
oled.DispChar("启动管理热点...", 0, 16)
oled.show()
try:
my_wifi.enable_APWiFi(essid=AP_SSID, password=AP_PASSWORD, channel=6)
try:
ip = my_wifi.ap.ifconfig()[0]
except Exception:
ip = "192.168.4.1"
except Exception as e:
print("AP启动失败:", e)
ip = "0.0.0.0"
print("智能生活主机模式:", wifi_mode)
print("设备地址: http://" + ip)
if wifi_mode == "AP":
print("连接热点:", AP_SSID, "密码:", AP_PASSWORD)
# ==================== UDP服务:接收从机广播 ====================
try:
slave_udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
slave_udp.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
except Exception:
pass
slave_udp.bind(("0.0.0.0", MASTER_UDP_PORT))
slave_udp.setblocking(False)
print("UDP主机已监听端口:", MASTER_UDP_PORT)
except Exception as e:
slave_udp = None
add_warning("UDP主从服务启动失败: %s" % str(e))
# ==================== HTTP服务器 ====================
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", 80))
server.listen(3)
server.setblocking(False)
last_oled_ms = 0
last_rfid_ms = 0
# ==================== 主循环 ====================
while True:
now = time.ticks_ms()
# 1) 收从机UDP心率和摔倒状态
poll_slave_udp()
# 2) P0人体感应:触发后同时打开P13/P14,超时后同时关闭
try:
pir_now = bool(pir_pin.read_digital()) if pir_pin is not None else False
except Exception:
pir_now = False
add_warning("P0人体感应读取异常")
if pir_now and pir_auto:
last_motion_ms = now
if not outputs["room_light"]:
set_room_lights(True)
if pir_auto and not pir_now:
if time.ticks_diff(now, last_motion_ms) >= light_timeout * 1000:
if outputs["room_light"]:
set_room_lights(False)
# 3) 每2秒读取SHT20
if time.ticks_diff(now, last_sensor_ms) >= 2000:
last_sensor_ms = now
try:
if sht20 is not None:
temperature = float(sht20.temperature())
humidity = float(sht20.humidity())
sensor_ok = True
else:
sensor_ok = False
except Exception as e:
sensor_ok = False
print("SHT20读取异常:", e)
# 4) 温湿度自动控制风扇:达到任一阈值开启,双低于回差阈值关闭
if (fan_auto and fan_available and sensor_ok and
temperature is not None and humidity is not None):
if not outputs["fan"]:
if temperature >= fan_temp_threshold or humidity >= fan_humidity_threshold:
set_fan(True)
else:
if (temperature <= fan_temp_threshold - 2.0 and
humidity <= fan_humidity_threshold - 5.0):
set_fan(False)
# 5) RFID卡号匹配后点亮P7门灯
if time.ticks_diff(now, last_rfid_ms) >= 250:
last_rfid_ms = now
try:
card = Scan_Rfid.scanning(i2c)
if card:
uid = int(card.serial_number())
if not rfid_card_present:
rfid_last = str(uid)
rfid_authorized = (uid == RFID_AUTHORIZED_UID)
if rfid_authorized:
set_door_light(True)
door_light_deadline = time.ticks_add(now, door_timeout * 1000)
print("RFID授权通过,门灯点亮,UID:", uid)
else:
print("RFID未授权,UID:", uid)
rfid_card_present = True
else:
rfid_card_present = False
except Exception as e:
rfid_available = False
add_warning("RFID读取异常,请检查I2C模块和bluebit驱动")
print("RFID扫描异常:", e)
if door_light_deadline is not None and time.ticks_diff(now, door_light_deadline) >= 0:
set_door_light(False)
door_light_deadline = None
# 6) 本地主机加速度计:摇晃时显示并上报疑似摔倒
detect_shake(now)
# 7) HTTP API请求
try:
conn, addr = server.accept()
handle_request(conn)
except OSError:
pass
except Exception as e:
print("HTTP服务器异常:", e)
# 8) OLED:报警状态优先显示,直到App清除
if time.ticks_diff(now, last_oled_ms) >= 500:
last_oled_ms = now
try:
oled.fill(0)
oled.DispChar("智能生活主机", 0, 0)
oled.DispChar("AP:" + AP_SSID if wifi_mode == "AP" else "IP:" + ip, 0, 16)
if fallen:
oled.DispChar("摔倒报警!!!", 0, 32)
oled.DispChar("请立即确认!", 0, 48)
else:
if sensor_ok and temperature is not None and humidity is not None:
oled.DispChar("T:%.1f H:%.0f%%" % (temperature, humidity), 0, 32)
else:
oled.DispChar("温湿度读取中", 0, 32)
hr_text = "HR:%s*" % (str(heart_rate_bpm) if heart_rate_bpm is not None else "--")
oled.DispChar(hr_text + (" UDP在线" if slave_connected else " UDP等待"), 0, 48)
oled.show()
except Exception:
pass
time.sleep_ms(20)
+351
View File
@@ -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)
Binary file not shown.

After

Width:  |  Height:  |  Size: 175 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 180 KiB

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 637 KiB

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 571 KiB

BIN
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 618 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 168 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 164 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 103 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 184 KiB