From 1d839f6e072aee98cd5c64b9fb127bc9d8ccb0f1 Mon Sep 17 00:00:00 2001 From: ZhouyiStudio Date: Fri, 30 Oct 2026 21:26:26 +0800 Subject: [PATCH] =?UTF-8?q?add:=20=E4=B8=8A=E4=BC=A0=E6=8E=8C=E6=8E=A7?= =?UTF-8?q?=E6=9D=BF=E4=B8=BB=E6=9C=BA=E7=A8=8B=E5=BA=8F=E6=BA=90=E7=A0=81?= =?UTF-8?q?main.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- onboardprogram/main.py | 709 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 709 insertions(+) create mode 100644 onboardprogram/main.py diff --git a/onboardprogram/main.py b/onboardprogram/main.py new file mode 100644 index 0000000..6566f9e --- /dev/null +++ b/onboardprogram/main.py @@ -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 = ( + "" + "" + "智能生活" + "

智能生活主机已启动

" + "

查看设备状态

" + "

Wi-Fi模式：%s

地址：%s

" + "" + ) % (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)