add: 上传掌控板主机程序源码main.py
This commit is contained in:
ZhouyiStudio committed
1 parent
d656c55ba7
commit
1d839f6e07
1 file changed
+709
@@ -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)
|
||||
Reference in new issue
Block a user