From 732feb57275b02ed70343c612478aa96822e2bde Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BA=AA=E7=94=B0=E9=87=8E?= Date: Thu, 21 May 2026 08:06:59 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=88=E7=89=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 472 ++++++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 390 insertions(+), 82 deletions(-) diff --git a/main.py b/main.py index a63c413..fa3d548 100644 --- a/main.py +++ b/main.py @@ -1,8 +1,15 @@ # ==================================================================================== # K210 V11 (分时网络通信) -# 功能更新: -# - 继承V10所有功能 -# - 【新功能】使用分时处理网络通信，每帧只做少量网络工作，避免长时间阻塞 +# 本文件运行在 Sipeed Maix/K210 的 MaixPy 环境中，是识别端的主程序。 +# 主要职责： +# 1. 通过摄像头采集画面，调用 KPU 模型完成人脸检测、关键点定位和特征提取； +# 2. 将识别结果通过 UART1 发送给 STM32，供 STM32 执行考勤判断、OLED 显示和蜂鸣器提示； +# 3. 通过 UART2 控制 ESP8266 连接服务端，上传新增人脸照片，同步学号映射和远程删除任务； +# 4. 将人脸特征、照片、face_id 与 student_id 映射持久化到 SD 卡，保证断电后仍可恢复。 +# 设计说明： +# - K210 的主任务是实时识别，因此网络通信不能长时间占用主循环； +# - 网络相关操作采用“分时处理”：每隔若干帧只处理一个小任务，尽量避免画面卡顿； +# - 新增人脸时先写入本地 SD 卡，再排队上传，防止网络异常导致照片或特征丢失。 # ==================================================================================== import sensor, image, lcd @@ -27,42 +34,61 @@ except: crypto_lib = None # --- 增加堆内存大小 --- -# K210有8MB SRAM，默认堆较小，增加到512KB用于图像压缩 +# K210 有 8MB SRAM，但 MaixPy 默认堆空间偏小。 +# 人脸照片压缩为 JPEG、HTTP 拼包、Base64/AES 加解密都会临时申请较大的内存， +# 因此启动时把 GC 堆扩大到 1MB，减少运行过程中因内存不足导致的压缩或网络失败。 utils.gc_heap_size(1024 * 1024) gc.collect() print("[BOOT] Heap size set to 1MB") # --- 配置参数（宏定义）--- -SYNC_CHECK_INTERVAL_MS = 50000 # 检查同步请求的时间间隔（毫秒） -NETWORK_FRAME_INTERVAL = 300 # 每N帧检查一次网络任务（约1秒） +SYNC_CHECK_INTERVAL_MS = 50000 # 检查服务端同步请求的时间间隔（毫秒），避免频繁访问接口 +NETWORK_FRAME_INTERVAL = 300 # 每 N 帧检查一次网络任务；数值越小同步越及时，但识别画面越容易卡顿 # --- 分时网络任务队列 --- -photo_upload_queue = [] # 照片上传队列 [face_id, ...]，JPEG先保存在SD卡 -network_task_pending = False # 是否有网络任务待处理 -frames_since_network = 0 # 距离上次网络操作的帧数 +photo_upload_queue = [] # 照片上传队列，保存待上传的 face_id；真正的 JPEG 数据存放在 SD 卡 +network_task_pending = False # 预留的网络任务状态位，目前主要由照片队列和同步标志驱动 +frames_since_network = 0 # 距离上次网络操作已经经过的帧数，用于在主循环中做分时调度 # --- 0. ESP8266 WiFi配置 --- -WIFI_SSID = "yzfh" # 修改为你的WiFi名称 -WIFI_PASSWORD = "yzfhroot" # 修改为你的WiFi密码 -SERVER_IP = "27.106.113.205" # 修改为服务器IP地址 -SERVER_PORT = 8082 # 服务器HTTP端口 -API_TOKEN = "atk_nKfVWLhMjL0RGSfRll3R_sdJS6IrIjuV" # 必须与服务端 ATTENDANCE_API_TOKEN 一致 -ESP_DEBUG = True # 打印ESP8266 AT响应，便于排查WiFi连接问题 -ESP_BAUD_RATES = (115200, 9600, 74880, 57600, 38400) -ESP_UART_BAUD = 115200 -ESP_BOOT_WAIT_MS = 6000 # 等待ESP8266上电启动和自动连网 -LOCAL_FACE_PHOTO_DIR = "/sd/face_photos" # 本地人脸照片缓存目录 +# 这些参数需要与实际部署环境保持一致： +# - WiFi 名称和密码决定 ESP8266 能否联网； +# - SERVER_IP/SERVER_PORT 指向考勤服务端； +# - API_TOKEN 必须与服务端 ATTENDANCE_API_TOKEN 一致，否则接口会拒绝请求。 +WIFI_SSID = "yzfh" # WiFi 名称 +WIFI_PASSWORD = "yzfhroot" # WiFi 密码 +SERVER_IP = "27.106.113.205" # 服务端 IP 地址 +SERVER_PORT = 8082 # 服务端 HTTP 端口 +API_TOKEN = "atk_nKfVWLhMjL0RGSfRll3R_sdJS6IrIjuV" # 接口访问令牌，和服务端配置保持一致 +ESP_DEBUG = True # 是否打印 ESP8266 AT 响应，调试连接问题时建议开启 +ESP_BAUD_RATES = (115200, 9600, 74880, 57600, 38400) # 常见 AT 固件波特率，启动时会自动探测 +ESP_UART_BAUD = 115200 # 当前使用的 ESP8266 串口波特率 +ESP_BOOT_WAIT_MS = 6000 # 等待 ESP8266 上电启动和自动联网的时间 +LOCAL_FACE_PHOTO_DIR = "/sd/face_photos" # 本地人脸照片缓存目录，照片上传失败时用于后续补传 + +# AES-CBC 加密参数。 +# K210 与服务端之间的 JSON 请求和响应使用 AES128-CBC + Base64 包装， +# 这样可以避免学号映射、删除任务等管理数据以明文形式在 HTTP 中传输。 AES_KEY = b"\x2b\x7e\x15\x16\x28\xae\xd2\xa6\xab\xf7\x15\x88\x09\xcf\x4f\x3c" AES_IV = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f" +# 函数作用：判断当前固件环境是否具备 AES 和 Base64 加解密能力。 def crypto_ready(): + # MaixPy 固件版本不同，可用的加密模块名称也不同。 + # 启动阶段已经兼容 ucryptolib/cryptolib，这里统一判断依赖是否齐全。 return crypto_lib is not None and ubinascii is not None +# 函数作用：为 AES-CBC 明文追加 PKCS#7 填充。 def pkcs7_pad(data): + # AES-CBC 要求输入长度必须是 16 字节的整数倍， + # PKCS#7 会在尾部补 N 个值为 N 的字节，服务端解密时按同一规则移除。 pad_len = 16 - (len(data) % 16) return data + bytes([pad_len]) * pad_len +# 函数作用：移除 AES-CBC 解密结果中的 PKCS#7 填充。 def pkcs7_unpad(data): + # 解密后移除 PKCS#7 填充。 + # 如果填充不合法，直接返回原始数据，避免因为异常数据导致主循环崩溃。 if not data: return b"" pad_len = data[-1] @@ -73,7 +99,10 @@ def pkcs7_unpad(data): return data return data[:-pad_len] +# 函数作用：根据 encrypt 参数执行 AES-CBC 加密或解密。 def aes_cbc_crypt(data, encrypt=True): + # encrypt=True 表示加密，False 表示解密。 + # MaixPy 的 aes(mode=2) 对应 CBC 模式，密钥和 IV 必须与服务端保持一致。 if not crypto_ready(): print("[CRYPTO] ucryptolib/ubinascii unavailable, refuse plaintext HTTP") return None @@ -84,7 +113,10 @@ def aes_cbc_crypt(data, encrypt=True): print("[CRYPTO] AES failed:", e) return None +# 函数作用：加密载荷并转换为 Base64 文本。 def encrypt_payload_base64(data): + # JSON 文本先转为 UTF-8 字节，再进行 PKCS#7 填充和 AES 加密， + # 最后用 Base64 转为可放进 HTTP body 的 ASCII 文本。 if isinstance(data, str): data = data.encode("utf-8") encrypted = aes_cbc_crypt(pkcs7_pad(data), True) @@ -96,7 +128,10 @@ def encrypt_payload_base64(data): except: return str(encoded, "utf-8") +# 函数作用：把 Base64 密文解码并解密为明文。 def decrypt_payload_base64(data): + # 服务端返回的加密 body 是 Base64 字符串。 + # 解密成功后返回 UTF-8 JSON 文本，失败时返回 None 由调用方决定是否重试。 if data is None: return None if isinstance(data, str): @@ -115,7 +150,10 @@ def decrypt_payload_base64(data): print("[CRYPTO] UTF-8 decode failed:", e) return None +# 函数作用：从 ESP8266 返回的 HTTP 响应中提取正文部分。 def extract_http_body(resp): + # ESP8266 透传回来的内容通常包含 AT 回显、HTTP 头、HTTP body、CLOSED/OK 等尾部文本。 + # 这里按 HTTP 头结束符拆出 body，并尽量依据 Content-Length 截断，避免把 CLOSED 当作 JSON。 if not resp: return "" marker = "\r\n\r\n" @@ -144,7 +182,10 @@ def extract_http_body(resp): body = body[:pos] return body.strip() +# 函数作用：判断 HTTP 响应正文是否已接收完整。 def http_body_complete(resp): + # 判断 HTTP 响应是否已经接收完整。 + # 有 Content-Length 时以长度为准；没有长度时只能退化为等待连接关闭。 if not resp or "\r\n\r\n" not in resp: return False body = resp.split("\r\n\r\n", 1)[1] @@ -162,7 +203,12 @@ def http_body_complete(resp): except: return "CLOSED" in resp +# 函数作用：提取并按需解密 HTTP JSON 响应正文。 def decode_http_json_body(resp): + # 统一处理服务端响应： + # - 带 X-Encrypted 响应头时按 AES 解密； + # - 明文 JSON 直接返回； + # - 其余情况尝试按加密 body 解析，兼容服务端响应头缺失的情况。 body = extract_http_body(resp) if not body: return None @@ -177,17 +223,25 @@ CAMERA_HMIRROR = 1 CAMERA_VFLIP = 1 # --- 1. UART通信初始化 --- -# UART1: 连接STM32（原有功能） +# UART1：连接 STM32。 +# K210 识别出人脸后，会发送类似： +# ID:20210001,NAME:20210001,CONF:0.95 +# 或： +# UNKNOWN +# STM32 端根据这条串口消息执行考勤记录、蜂鸣器和 OLED 显示逻辑。 fm.register(24, fm.fpioa.UART1_TX, force=True) fm.register(25, fm.fpioa.UART1_RX, force=True) uart = UART(UART.UART1, 115200, 8, 0, 1, timeout=1000, read_buf_len=4096) -# UART2: 连接新ESP8266（照片上传功能） +# UART2：连接 ESP8266。 +# 该串口只负责网络通信：WiFi 连接、HTTP GET/POST、照片上传和同步请求。 +# read_buf_len 设置为 4096 是为了尽量一次容纳较长的 HTTP 响应片段。 fm.register(34, fm.fpioa.UART2_TX, force=True) fm.register(35, fm.fpioa.UART2_RX, force=True) uart2 = UART(UART.UART2, 115200, 8, 0, 1, timeout=2000, read_buf_len=4096) # --- ESP8266 WiFi通信函数（优化版 - 减少阻塞）--- +# 函数作用：按指定波特率重新打开 ESP8266 串口。 def esp_reopen_uart(baud): """按指定波特率重新打开ESP8266串口""" global uart2, ESP_UART_BAUD @@ -199,17 +253,23 @@ def esp_reopen_uart(baud): uart2 = UART(UART.UART2, baud, 8, 0, 1, timeout=2000, read_buf_len=4096) utime.sleep_ms(100) +# 函数作用：清空 ESP8266 串口接收缓冲区。 def esp_clear_rx(): """清空ESP8266串口残留数据""" + # AT 固件经常会在命令之外输出 ready、WIFI GOT IP、CLOSED 等异步文本。 + # 每次发送新命令前先清空缓冲区，能减少上一条响应污染本次解析的概率。 while uart2.any(): uart2.read() utime.sleep_ms(10) +# 函数作用：打印 ESP8266 指令响应日志，便于串口调试。 def esp_log_response(cmd, response): + # 调试日志只截取前 180 个字符，避免长 HTTP 响应刷屏影响串口观察。 if ESP_DEBUG: text = response.replace('\r\n', ' ').strip() if response else "(空)" print("[ESP] {} => {}".format(cmd, text[:180])) +# 函数作用：向 ESP8266 发送 AT 指令并等待期望响应。 def esp_send_cmd(cmd, timeout=1000, expected=None): """发送AT命令到ESP8266并等待响应""" esp_clear_rx() @@ -219,6 +279,8 @@ def esp_send_cmd(cmd, timeout=1000, expected=None): response = "" if expected is None: expected = ("OK", "ERROR", "FAIL", "ready") + # 轮询串口直到出现期望关键字或超时。 + # 这里不用阻塞式长等待，是为了在网络异常时尽快回到识别主循环。 while utime.ticks_diff(utime.ticks_ms(), start) < timeout: if uart2.any(): data = uart2.read() @@ -230,8 +292,11 @@ def esp_send_cmd(cmd, timeout=1000, expected=None): esp_log_response(cmd, response) return response +# 函数作用：探测 ESP8266 串口是否能正常响应 AT 指令。 def esp_probe_uart(): """探测ESP8266实际AT波特率""" + # 不同 ESP8266 AT 固件默认波特率可能不同；如果 115200 没有响应， + # 按常见波特率逐个尝试，找到能返回 OK 的配置后继续使用。 print("[ESP] Probing UART baud rate...") for baud in ESP_BAUD_RATES: print("[ESP] Try baud:", baud) @@ -246,6 +311,7 @@ def esp_probe_uart(): print("[ESP] No AT response on tested baud rates") return False +# 函数作用：等待 ESP8266 上电启动完成。 def esp_wait_boot(): """等待ESP8266启动完成，避免K210太早发送AT命令""" print("[ESP] Waiting for module boot...") @@ -261,17 +327,22 @@ def esp_wait_boot(): print("[ESP] Boot log:", boot_log.replace('\r\n', ' ')[:180]) esp_clear_rx() +# 函数作用：检查 ESP8266 当前 WiFi 连接状态。 def esp_check_wifi_status(): """检查WiFi连接状态（快速检查）""" + # AT+CWJAP? 会返回当前连接的 AP 名称。 + # 只做轻量检查，不重新连接，适合在上传照片前快速确认网络状态。 resp = esp_send_cmd("AT+CWJAP?", 1500) return WIFI_SSID in resp +# 函数作用：控制 ESP8266 连接配置好的 WiFi 热点。 def esp_connect_wifi(): """连接WiFi（单次尝试）""" print("[ESP] Connecting to WiFi:", WIFI_SSID) - resp = esp_send_cmd("ATE0", 1000) + resp = esp_send_cmd("ATE0", 1000) # 关闭命令回显，后续解析响应更干净 + # 使用 station 模式连接路由器。部分 AT 固件支持 _CUR 命令，部分只支持普通命令。 resp = esp_send_cmd("AT+CWMODE_CUR=1", 1500) if "OK" not in resp: resp = esp_send_cmd("AT+CWMODE=1", 1500) @@ -284,7 +355,8 @@ def esp_connect_wifi(): print("[ESP] WiFi already connected!") return True - # 连接WiFi。部分路由器获取IP较慢，等待时间不能太短。 + # 连接 WiFi。部分路由器 DHCP 分配 IP 较慢，等待时间不能太短。 + # 先尝试 AT+CJAP_CUR，失败后再回退到 AT+CWJAP，兼容不同 AT 固件版本。 cmd = 'AT+CWJAP_CUR="{}","{}"'.format(WIFI_SSID, WIFI_PASSWORD) resp = esp_send_cmd(cmd, 20000, ("WIFI GOT IP", "OK", "FAIL", "ERROR")) @@ -305,13 +377,14 @@ def esp_connect_wifi(): print("[ESP] WiFi connect failed") return False +# 函数作用：初始化 ESP8266 并尝试连接 WiFi。 def esp_init_wifi(max_retries=5): """初始化ESP8266并连接WiFi（减少重试次数）""" print("[ESP] Initializing...") esp_wait_boot() esp_clear_rx() - # 测试AT响应 + # 测试 AT 响应。若默认波特率没有响应，再调用 esp_probe_uart 自动探测。 for i in range(max_retries): resp = esp_send_cmd("AT", 2500) if "OK" in resp: @@ -324,7 +397,7 @@ def esp_init_wifi(max_retries=5): print("[ESP] ESP8266 not responding!") return False - # 连接WiFi + # 连接 WiFi。初始化阶段允许多次重试，运行中重连时通常只传入 1 次，避免阻塞太久。 for i in range(max_retries): if esp_connect_wifi(): return True @@ -332,6 +405,7 @@ def esp_init_wifi(max_retries=5): return False +# 函数作用：确保 WiFi 连接可用，断开时自动重连。 def esp_ensure_wifi(): """确保WiFi已连接（快速检查）""" global wifi_connected @@ -340,8 +414,11 @@ def esp_ensure_wifi(): return wifi_connected return wifi_connected +# 函数作用：确保本地人脸照片缓存目录存在。 def ensure_local_photo_dir(): """确保SD卡本地照片目录存在""" + # MaixPy 的 os.mkdir 只能创建单级目录；这里目录路径固定为 /sd/face_photos， + # 启动或新增人脸时确认一次即可。 try: os.listdir(LOCAL_FACE_PHOTO_DIR) return True @@ -354,11 +431,16 @@ def ensure_local_photo_dir(): print("[PHOTO] Create local dir failed:", e) return False +# 函数作用：生成指定人脸编号的本地照片路径。 def local_face_photo_path(face_id): + # 使用 face_id 作为照片文件名，保证照片、特征和服务端档案可以稳定对应。 return "{}/{}.jpg".format(LOCAL_FACE_PHOTO_DIR, face_id) +# 函数作用：把采集到的人脸照片保存到本地 SD 卡。 def save_local_face_photo(face_id, jpeg_data): """先把人脸照片保存到SD卡，保证本地特征和照片一致""" + # 新增人脸时必须先保存照片再写入上传队列。 + # 这样即使 ESP8266 掉线，下一次启动仍能扫描 SD 卡并补传这张照片。 if not ensure_local_photo_dir(): return False path = local_face_photo_path(face_id) @@ -371,7 +453,9 @@ def save_local_face_photo(face_id, jpeg_data): print("[PHOTO] Save local photo failed:", e) return False +# 函数作用：读取指定人脸编号的本地照片数据。 def load_local_face_photo(face_id): + # 上传队列只保存 face_id，真正上传前再从 SD 卡读取 JPEG，减少常驻内存占用。 path = local_face_photo_path(face_id) try: with open(path, "rb") as f: @@ -380,12 +464,16 @@ def load_local_face_photo(face_id): print("[PHOTO] Read local photo failed {}: {}".format(path, e)) return None +# 函数作用：从待上传照片队列中移除指定人脸编号。 def remove_photo_from_queue(face_id): + # 删除人脸或上传完成后，把对应 face_id 从待上传队列中移除，避免重复上传。 global photo_upload_queue face_id = str(face_id) photo_upload_queue = [x for x in photo_upload_queue if x != face_id] +# 函数作用：删除指定人脸编号的本地照片缓存。 def delete_local_face_photo(face_id): + # 本地照片与本地特征是一组数据；删除人脸时也删除照片，避免服务端之后补传无效照片。 face_id = str(face_id) remove_photo_from_queue(face_id) path = local_face_photo_path(face_id) @@ -396,7 +484,9 @@ def delete_local_face_photo(face_id): except: return False +# 函数作用：删除所有本地人脸照片缓存。 def delete_all_local_face_photos(): + # 清空全部人脸时同步清理照片目录和上传队列，保持 SD 卡状态一致。 if not ensure_local_photo_dir(): return try: @@ -411,13 +501,20 @@ def delete_all_local_face_photos(): except Exception as e: print("[PHOTO] Delete all local photos failed:", e) +# 函数作用：通过 ESP8266 发送一段原始 TCP 数据。 def esp_send_data(data, timeout=1500): """发送数据并等待SEND OK（优化超时）""" + # ESP8266 AT 发送流程固定为： + # 1. AT+CIPSEND=<长度> + # 2. 等待 '>' 提示符 + # 3. 写入数据 + # 4. 等待 SEND OK + # 这里封装成一个函数，供 HTTP 头、JSON body、JPEG 分块共用。 cmd = "AT+CIPSEND={}".format(len(data)) uart2.write(cmd + "\r\n") utime.sleep_ms(50) - # 等待 > 提示符（缩短超时） + # 等待 '>' 提示符。若没有提示符，说明 TCP 连接不可写，直接失败。 start = utime.ticks_ms() resp = "" got_prompt = False @@ -434,10 +531,10 @@ def esp_send_data(data, timeout=1500): if not got_prompt: return False - # 发送数据 + # 发送实际数据。data 既可能是字符串，也可能是 JPEG bytes。 uart2.write(data) - # 等待 SEND OK（缩短超时） + # 等待 SEND OK。超时设置偏短，是为了网络异常时不长时间卡住识别。 start = utime.ticks_ms() resp = "" while utime.ticks_diff(utime.ticks_ms(), start) < timeout: @@ -451,26 +548,27 @@ def esp_send_data(data, timeout=1500): return False +# 函数作用：通过 HTTP POST 上传指定人脸照片到服务端。 def esp_http_post_photo(face_id, jpeg_data, max_retries=2): """通过ESP8266发送HTTP POST请求上传照片""" - # 直接使用face_id作为照片ID（不加时间戳） + # 直接使用 face_id 作为照片 ID（不加时间戳），服务端可用该 ID 覆盖或更新同一张人脸照片。 print("[ESP] Uploading photo for face ID:", face_id, "size:", len(jpeg_data)) for attempt in range(max_retries): - # 确保WiFi连接 + # 上传前确认 WiFi 已连接；若未连接则尝试一次快速重连。 if not esp_ensure_wifi(): print("[ESP] No WiFi, retry {}/{}".format(attempt+1, max_retries)) utime.sleep_ms(2000) continue - # 关闭之前的连接 + # 关闭之前残留的 TCP 连接，避免 CIPSEND 写入旧连接导致失败。 esp_send_cmd("AT+CIPCLOSE", 500) utime.sleep_ms(100) - # 设置单连接模式 + # 设置单连接模式，简化后续 CIPSEND 和关闭连接的处理。 esp_send_cmd("AT+CIPMUX=0", 500) - # 建立TCP连接 + # 建立到服务端 HTTP 端口的 TCP 连接。 cmd = 'AT+CIPSTART="TCP","{}",{}'.format(SERVER_IP, SERVER_PORT) resp = esp_send_cmd(cmd, 3000) if "OK" not in resp and "CONNECT" not in resp: @@ -480,7 +578,9 @@ def esp_http_post_photo(face_id, jpeg_data, max_retries=2): print("[ESP] TCP connected, sending data...") - # 使用face_id作为文件名 + # 构造照片上传请求。 + # 这里 Content-Type 使用 image/jpeg，body 直接放 JPEG bytes，不参与 AES 加密， + # 因为照片本身较大，K210 端加密会显著增加内存压力和上传时间。 http_header = "POST /api/v1/faces/photos/{} HTTP/1.1\r\n".format(face_id) http_header += "Host: {}:{}\r\n".format(SERVER_IP, SERVER_PORT) http_header += "Content-Type: image/jpeg\r\n" @@ -488,13 +588,14 @@ def esp_http_post_photo(face_id, jpeg_data, max_retries=2): http_header += "Content-Length: {}\r\n".format(len(jpeg_data)) http_header += "Connection: close\r\n\r\n" - # 发送HTTP头 + # 先发送 HTTP 头。照片 body 后续分块发送。 if not esp_send_data(http_header): print("[ESP] Send header failed") esp_send_cmd("AT+CIPCLOSE", 500) continue - # 分块发送JPEG数据 + # 分块发送 JPEG 数据。 + # ESP8266 AT 透传缓冲区有限，大照片一次性发送容易失败，按 1024 字节拆分更稳。 CHUNK_SIZE = 1024 offset = 0 send_ok = True @@ -513,7 +614,8 @@ def esp_http_post_photo(face_id, jpeg_data, max_retries=2): utime.sleep_ms(500) continue - # 等待服务器响应 + # 等待服务器响应。这里只要数据全部 SEND OK，基本可认为上传成功； + # 响应读取主要用于调试和及时发现服务端拒绝。 print("[ESP] Waiting for response...") resp = "" start = utime.ticks_ms() @@ -541,19 +643,21 @@ def esp_http_post_photo(face_id, jpeg_data, max_retries=2): print("[ESP] Upload failed after {} retries".format(max_retries)) return False +# 函数作用：保存当前人脸照片并尝试上传到服务端。 def capture_and_upload_face(face_id, img): """拍摄人脸照片并加入上传队列（非阻塞）""" try: - # 先复制图像，避免修改原图 + # 先复制图像，避免压缩、转换等操作影响当前 LCD 显示和后续识别处理。 img_copy = img.copy() - # 转换为RGB565格式（如果需要）后压缩为JPEG + # 将当前画面压缩为 JPEG。 + # quality=50 在清晰度和体积之间折中，适合 ESP8266 低速串口上传。 try: jpeg_img = img_copy.compress(quality=50) # 降低质量减小体积 jpeg_data = jpeg_img.to_bytes() except Exception as e1: print("[PHOTO] Compress method 1 failed:", e1) - # 尝试另一种方式 + # 如果当前固件不支持 compress，则尝试直接取字节，保留兼容性。 try: jpeg_data = img_copy.to_bytes() # 直接尝试转换 except Exception as e2: @@ -566,7 +670,8 @@ def capture_and_upload_face(face_id, img): print("[PHOTO] JPEG too small, skip upload") return False - # 先保存本地，再加入上传队列（非阻塞）。上传失败也不会丢照片。 + # 先保存本地，再加入上传队列（非阻塞）。 + # 注意：这里不直接调用 HTTP 上传，避免新增人脸时 UI 长时间卡住。 if not save_local_face_photo(face_id, jpeg_data): return False @@ -578,7 +683,8 @@ def capture_and_upload_face(face_id, img): print("[PHOTO] Error:", e) return False -# 初始化WiFi连接（启动时尝试连接） +# 初始化 WiFi 连接（启动时尝试连接）。 +# 若启动时失败，识别功能仍然可用；照片会保存在 SD 卡，等后续重启或网络恢复后补传。 wifi_connected = False print("[BOOT] Starting ESP8266 WiFi init...") try: @@ -588,12 +694,19 @@ except Exception as e: print("[BOOT] ESP8266 init error:", e) # --- 人脸ID到真实学号的映射表 --- +# K210 内部识别使用的是本地 face_id，例如 ID:1、ID:2。 +# 管理端和考勤记录更关心真实学号，因此单独维护 face_id -> student_id 的映射。 +# 这样可以做到： +# - recordftr.txt 只保存 K210 原始人脸特征和 face_id，不因学生信息变更而改动特征库； +# - 管理端可以远程调整学号绑定，K210 同步后立刻按新学号上报给 STM32； +# - 未绑定学号的人脸仍可使用 face_id 作为兜底身份，便于调试和后续补绑定。 face_id_mapping = {} # 格式: {"1": "20210001", "2": "20210002"} -MAPPING_FILE = "/sd/face_mapping.txt" # SD卡上的映射文件 -SYNC_VERSION_FILE = "/sd/sync_version.txt" # 本地已应用的服务端同步版本 +MAPPING_FILE = "/sd/face_mapping.txt" # SD 卡上的映射文件，简单使用 key=value 文本格式 +SYNC_VERSION_FILE = "/sd/sync_version.txt" # 本地已应用的服务端同步版本号 RECORD_PATH = "/sd/recordftr.txt" # 人脸特征记录文件路径 -sync_last_version = 0 +sync_last_version = 0 # K210 已处理到的服务端统一同步版本 +# 函数作用：从本地文件加载人脸编号到学号的映射关系。 def load_local_mapping(): """从SD卡加载本地映射配置""" global face_id_mapping @@ -603,6 +716,8 @@ def load_local_mapping(): for line in f: line = line.strip() if '=' in line: + # 每行格式为 face_id=student_id。 + # 不使用 JSON 是为了降低 MaixPy 端解析复杂度，并方便现场手动查看。 parts = line.split('=') if len(parts) == 2: face_id_mapping[parts[0]] = parts[1] @@ -610,6 +725,7 @@ def load_local_mapping(): except: print("[LOCAL] No local mapping file") +# 函数作用：把当前人脸编号到学号映射保存到本地文件。 def save_local_mapping(): """保存映射配置到SD卡""" try: @@ -622,8 +738,11 @@ def save_local_mapping(): print("[LOCAL] Save error:", e) return False +# 函数作用：从本地文件读取最近处理过的服务端同步版本号。 def load_sync_version(): """从SD卡加载本地已同步版本""" + # 服务端每次变更人脸档案或删除任务都会推进统一版本号。 + # K210 保存最后一次成功应用的版本，下次只拉取增量状态，避免重复处理旧任务。 global sync_last_version try: with open(SYNC_VERSION_FILE, "r") as f: @@ -633,8 +752,11 @@ def load_sync_version(): sync_last_version = 0 print("[LOCAL] Sync version:", sync_last_version) +# 函数作用：保存最近处理过的服务端同步版本号。 def save_sync_version(version): """保存本地已同步版本""" + # 只有在映射应用、删除任务和 ACK 都成功后才推进版本号。 + # 如果删除失败则保留旧版本，下一轮同步会继续重试。 global sync_last_version sync_last_version = version try: @@ -643,12 +765,15 @@ def save_sync_version(version): except Exception as e: print("[LOCAL] Save sync version error:", e) +# 函数作用：更新本地人脸映射记录中的学生编号。 def update_face_record(face_id, new_student_id): """保存face_id到学号的映射，保持recordftr.txt中的原始face_id不变。""" global face_id_mapping, peoples face_id = str(face_id) new_student_id = str(new_student_id) + # 服务端可能存在历史映射或照片占位，但 K210 本地没有对应特征。 + # 这类映射不能直接保存，否则识别端会显示不存在的人脸档案。 if not local_face_id_exists(face_id): print("[UPDATE] Ignore mapping for missing local face_id {} -> {}".format(face_id, new_student_id)) return False @@ -656,6 +781,8 @@ def update_face_record(face_id, new_student_id): face_id_mapping[face_id] = new_student_id save_local_mapping() + # 只验证本地特征是否存在，不修改 recordftr.txt 中的人脸名称。 + # 识别算法仍使用原始 ID:x，串口发送前再查映射转换成学号。 found = False for p in peoples: if p.name == "ID:{}".format(face_id) or p.name == face_id: @@ -668,6 +795,7 @@ def update_face_record(face_id, new_student_id): print("[UPDATE] Mapping saved, local face feature not found yet: {}".format(face_id)) return True +# 函数作用：通过 ESP8266 发起 HTTP GET 请求并返回响应。 def esp_http_get(url_path, timeout=8000): """发送HTTP GET请求并返回响应内容""" if not wifi_connected: @@ -693,17 +821,18 @@ def esp_http_get(url_path, timeout=8000): print("[HTTP_GET] TCP连接成功") utime.sleep_ms(200) - # 构建HTTP GET请求 + # 构建 HTTP GET 请求。 + # X-Encrypted-Response 表示希望服务端返回 AES 加密后的 JSON，保护同步数据。 http_request = "GET {} HTTP/1.1\r\nHost: {}:{}\r\nX-Api-Token: {}\r\nX-Encrypted-Response: true\r\nConnection: close\r\n\r\n".format( url_path, SERVER_IP, SERVER_PORT, API_TOKEN) - # 使用AT+CIPSEND发送 + # 使用 AT+CIPSEND 发送请求头。GET 请求没有 body，所以一次发送即可。 print("[HTTP_GET] 准备发送 {} bytes".format(len(http_request))) cmd = "AT+CIPSEND={}".format(len(http_request)) uart2.write(cmd + "\r\n") utime.sleep_ms(100) - # 等待 > 提示符 + # 等待 '>' 提示符，说明 ESP8266 已准备接收待发送数据。 start = utime.ticks_ms() prompt_resp = "" while utime.ticks_diff(utime.ticks_ms(), start) < 2000: @@ -722,10 +851,10 @@ def esp_http_get(url_path, timeout=8000): print("[HTTP_GET] 收到>提示符，发送请求...") - # 发送HTTP请求 + # 发送 HTTP 请求文本。 uart2.write(http_request) - # 等待SEND OK + # 等待 SEND OK，确认 ESP8266 已把请求写入 TCP 连接。 start = utime.ticks_ms() send_resp = "" while utime.ticks_diff(utime.ticks_ms(), start) < 3000: @@ -743,7 +872,8 @@ def esp_http_get(url_path, timeout=8000): esp_send_cmd("AT+CIPCLOSE", 300) return None - # 等待服务器响应（关键：要等足够长的时间） + # 等待服务器响应。 + # 同步接口返回内容可能较长，所以不能只等第一次串口可读就结束。 print("[HTTP_GET] 等待服务器响应...") resp = send_resp # 继续累积数据 start = utime.ticks_ms() @@ -786,29 +916,39 @@ def esp_http_get(url_path, timeout=8000): print("[HTTP_GET] 未找到可用JSON") return None +# 函数作用：从人脸名称中提取纯人脸编号。 def extract_face_id(face_name): + # recordftr.txt 中的人脸名称一般是 ID:数字。 + # 这里统一剥离 ID: 前缀，后续映射、删除和照片文件名都使用纯数字 face_id。 face_name = str(face_name) if face_name.startswith("ID:"): return face_name.split(":", 1)[1] return face_name +# 函数作用：判断本地人脸库中是否存在指定人脸编号。 def local_face_id_exists(face_id): + # 判断本地特征库中是否存在该 face_id，避免服务端旧数据覆盖本地状态。 face_id = str(face_id) for p in peoples: if extract_face_id(p.name) == face_id: return True return False +# 函数作用：根据人脸名称获取映射后的真实学号。 def get_mapped_student_id(face_name): + # 根据识别到的人脸名称查找真实学号。 + # 如果没有绑定，返回空字符串，由调用方决定是否使用 face_id 兜底。 face_id = extract_face_id(face_name) return face_id_mapping.get(face_id, "") +# 函数作用：获取用于屏幕显示的姓名或学号文本。 def get_display_name(face_name): """显示学号映射；未绑定时显示原始face_id。""" mapped = get_mapped_student_id(face_name) return mapped if mapped else face_name -# 加载本地映射配置 +# 加载本地映射配置和同步版本。 +# 这一步必须在同步逻辑前执行，否则无法判断服务端状态是否比本地更新。 load_local_mapping() load_sync_version() @@ -816,14 +956,16 @@ load_sync_version() # --- 定时同步配置 --- last_sync_time = utime.ticks_ms() -sync_in_progress = False # 同步状态标志 -sync_step = 0 # 同步步骤（分步执行避免长时间阻塞） +sync_in_progress = False # 预留同步状态标志，便于后续改成更细粒度的分步同步 +sync_step = 0 # 预留同步步骤编号，当前版本的同步仍以一次接口调用为主 # --- 远程删除人脸功能 --- +# 函数作用：按人脸编号删除本地人脸记录及对应缓存。 def delete_face_by_id(face_id): """根据人脸ID删除本地人脸数据""" global peoples + # 本地特征库以 ID:face_id 命名，因此远程删除任务只需要传入纯 face_id。 target_patterns = ["ID:{}".format(face_id)] print("[DELETE] Looking for patterns:", target_patterns) @@ -841,6 +983,7 @@ def delete_face_by_id(face_id): deleted_name = peoples[deleted_index].name peoples.pop(deleted_index) try: + # 删除内存中的特征后立刻重写 recordftr.txt，保证下次启动不会恢复已删除人脸。 write_face_record() print("[DELETE] 删除人脸成功: {}".format(deleted_name)) @@ -856,12 +999,12 @@ def delete_face_by_id(face_id): else: delete_local_face_photo(pattern) - # 同时删除映射 + # 同时删除学号映射，避免 UI 或串口发送仍显示已删除人脸的绑定关系。 if face_id in face_id_mapping: del face_id_mapping[face_id] save_local_mapping() - # 重新加载 + # 重新加载特征库，确认文件和内存状态一致。 load_faces() return True except Exception as e: @@ -872,10 +1015,12 @@ def delete_face_by_id(face_id): print("[DELETE] 未找到人脸ID: {}".format(face_id)) return False +# 函数作用：通过 ESP8266 发送加密 JSON POST 请求。 def esp_http_post_json(url_path, json_data, timeout=2000): """发送HTTP POST请求（优化版）""" if not wifi_connected: return None + # JSON 管理数据统一加密后再发送，HTTP body 使用 application/octet-stream。 encrypted_data = encrypt_payload_base64(json_data) if encrypted_data is None: return None @@ -889,6 +1034,7 @@ def esp_http_post_json(url_path, json_data, timeout=2000): if "OK" not in resp and "CONNECT" not in resp: return None + # 请求头同时声明本次请求体已加密，并要求服务端响应也以加密形式返回。 http_request = "POST {} HTTP/1.1\r\nHost: {}:{}\r\nContent-Type: application/octet-stream\r\nX-Api-Token: {}\r\nX-Encrypted: AES128-CBC\r\nX-Encrypted-Response: true\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}".format( url_path, SERVER_IP, SERVER_PORT, API_TOKEN, len(encrypted_data), encrypted_data) @@ -910,7 +1056,10 @@ def esp_http_post_json(url_path, json_data, timeout=2000): esp_send_cmd("AT+CIPCLOSE", 300) return decode_http_json_body(resp) +# 函数作用：从简单 JSON 文本中读取整数字段。 def json_get_int(text, key, default=0): + # MaixPy 环境下完整 JSON 库可能不可用或占用内存较多。 + # 服务端返回结构固定，因此用轻量字符串扫描提取简单整数字段。 marker = '"{}":'.format(key) start = text.find(marker) if start == -1: @@ -926,7 +1075,10 @@ def json_get_int(text, key, default=0): except: return default +# 函数作用：从 JSON 文本中截取指定字段对应的对象或数组片段。 def extract_json_block(text, key, open_ch, close_ch): + # 从固定 JSON 文本中提取数组或对象片段，例如 face_profiles: [...]。 + # 使用括号深度计数，避免数组内部对象的右括号提前截断。 marker = '"{}":'.format(key) pos = text.find(marker) if pos == -1: @@ -945,7 +1097,9 @@ def extract_json_block(text, key, open_ch, close_ch): return text[start:i + 1] return "" +# 函数作用：解析 JSON 字符串数组片段。 def parse_string_array_block(arr_str): + # 解析 ["1","2"] 这类简单字符串数组，用于 pending_delete 列表。 values = [] pos = 0 while pos < len(arr_str): @@ -959,7 +1113,10 @@ def parse_string_array_block(arr_str): pos = end + 1 return values +# 函数作用：解析服务端下发的待删除人脸任务列表。 def parse_delete_tasks(tasks_str): + # 解析删除任务数组。每个任务包含 task_id 和 face_id； + # task_id 用于服务端 ACK，face_id 用于删除本地特征、照片和映射。 tasks = [] pos = 0 while pos < len(tasks_str): @@ -977,7 +1134,10 @@ def parse_delete_tasks(tasks_str): pos = obj_end + 1 return tasks +# 函数作用：从简单 JSON 文本中读取字符串字段。 def json_get_string(text, key, default=""): + # 从固定 JSON 对象中提取字符串字段。 + # 该函数不处理复杂转义字符，适用于当前服务端返回的简单 ID/状态字段。 marker = '"{}":"'.format(key) start = text.find(marker) if start == -1: @@ -988,6 +1148,7 @@ def json_get_string(text, key, default=""): return default return text[start:end] +# 函数作用：从服务端照片列表响应中解析人脸照片编号。 def parse_photo_ids(response): """解析服务端 /api/v1/faces/photos 返回的照片ID列表""" ids = {} @@ -1007,7 +1168,10 @@ def parse_photo_ids(response): pos = end + 1 return ids +# 函数作用：解析 JSON 对象片段为本地映射字典。 def parse_mapping_object(obj_str): + # 解析 {"1":"20210001","2":"20210002"} 形式的简单映射对象。 + # 当前主要保留作兼容工具，统一同步逻辑优先使用 face_profiles 数组。 mappings = {} if len(obj_str) < 2: return mappings @@ -1025,23 +1189,33 @@ def parse_mapping_object(obj_str): mappings[key] = value return mappings +# 函数作用：应用服务端人脸档案映射并更新本地记录。 def apply_face_profiles(server_mappings): + # 应用服务端下发的人脸档案映射。 + # server_mappings 的 key 是 face_id，value 是 student_id； + # 特殊值 "__PHOTO_ONLY__" 表示服务端只有照片占位，不应覆盖本地已有映射。 global face_id_mapping print("[SYNC_STATE] 服务器映射数量: {}".format(len(server_mappings))) updated_count = 0 valid_mappings = {} + # 先保留本地仍然有效的映射，顺手丢弃本地已经没有特征的人脸绑定。 for face_id, student_id in face_id_mapping.items(): if local_face_id_exists(face_id) and student_id: valid_mappings[face_id] = student_id for face_id, student_id in server_mappings.items(): + # 服务端存在但本地没有特征的人脸，不写入映射。 + # 这样可以避免管理端的旧照片或历史数据影响当前识别端。 if not local_face_id_exists(face_id): print("[SYNC_STATE] 跳过本地不存在的face_id映射: ID:{} -> {}".format(face_id, student_id)) continue if student_id == "__PHOTO_ONLY__": + # 照片占位通常来自先上传照片、后在管理端绑定学生的流程。 + # 此时不清空本地映射，等待真正 student_id 下发。 print("[SYNC_STATE] 服务端仅有照片占位，保留本地映射: ID:{}".format(face_id)) continue if not student_id: + # 服务端显式下发空 student_id，表示解除绑定。 if face_id in valid_mappings: del valid_mappings[face_id] updated_count += 1 @@ -1055,6 +1229,7 @@ def apply_face_profiles(server_mappings): save_local_mapping() print("[SYNC_STATE] 映射应用完成，更新 {} 条".format(updated_count)) +# 函数作用：从服务端同步统一状态，包括映射、照片和删除任务。 def sync_unified_state(): """按服务端统一版本号同步K210所需状态""" global sync_last_version @@ -1062,6 +1237,10 @@ def sync_unified_state(): print("[SYNC_STATE] WiFi未连接，跳过") return False + # 统一同步接口根据 client 和 last_version 返回 K210 需要处理的所有变化： + # - face_profiles：人脸与学号绑定关系； + # - delete_tasks/pending_delete：需要本地删除的人脸； + # - version/changed：服务端当前版本和是否有变化。 path = "/api/v1/sync/state?client=k210&last_version={}".format(sync_last_version) print("[SYNC_STATE] GET {}".format(path)) response = esp_http_get(path, 8000) @@ -1074,6 +1253,7 @@ def sync_unified_state(): print("[SYNC_STATE] local={}, server={}, changed={}".format(sync_last_version, server_version, changed)) if not changed: + # 服务端没有变化也保存版本号，防止本地版本落后但无需实际处理。 save_sync_version(server_version) return True @@ -1093,6 +1273,8 @@ def sync_unified_state(): student_id = json_get_string(obj, "student_id", "") status = json_get_string(obj, "status", "") if face_id: + # status=photo_only 表示该 face_id 在服务端只有照片，没有绑定学号。 + # 用特殊占位值传给 apply_face_profiles，避免误删本地已有绑定。 server_face_ids[face_id] = 1 if status == "photo_only" and not student_id: student_id = "__PHOTO_ONLY__" @@ -1104,6 +1286,7 @@ def sync_unified_state(): pending_arr = extract_json_block(response, "pending_delete", '[', ']') pending_delete = parse_string_array_block(pending_arr) if not delete_tasks: + # 兼容旧接口：只有 pending_delete 字符串数组时，临时转换成统一任务格式。 delete_tasks = [{"task_id": "", "face_id": face_id} for face_id in pending_delete] deleted_ids = [] deleted_task_ids = [] @@ -1113,12 +1296,15 @@ def sync_unified_state(): for task in delete_tasks: face_id = task["face_id"] if face_id in server_face_ids: + # 如果同一轮响应里仍然包含该 face_id 的服务端档案，说明这是旧删除任务。 + # 这时不删除本地人脸，但仍 ACK，防止服务端重复下发。 print("[SYNC_STATE] 忽略旧删除任务，服务端仍有人脸记录: {}".format(face_id)) deleted_ids.append(face_id) if task["task_id"]: deleted_task_ids.append(task["task_id"]) continue if not local_face_id_exists(face_id): + # 本地已经没有这张脸，等价于删除完成，直接 ACK。 print("[SYNC_STATE] 删除任务face_id本地不存在，跳过但ACK: {}".format(face_id)) deleted_ids.append(face_id) if task["task_id"]: @@ -1132,6 +1318,8 @@ def sync_unified_state(): delete_failed = True if deleted_ids: + # 删除完成后向服务端 ACK。只有 ACK 成功才推进本地版本， + # 否则下次同步继续重试，避免服务端和 K210 状态不一致。 deleted_json = ",".join(['"{}"'.format(x) for x in deleted_ids]) task_json = ",".join(['"{}"'.format(x) for x in deleted_task_ids]) ack_body = '{{"client":"k210","version":{},"deleted":[{}],"task_ids":[{}]}}'.format(server_version, deleted_json, task_json) @@ -1143,6 +1331,7 @@ def sync_unified_state(): delete_failed = True print("[SYNC_STATE] 删除ACK失败，暂不推进本地版本，等待下次重试") + # 删除任务处理完成后再应用映射，避免先写入后又删除造成短暂脏状态。 apply_face_profiles(server_mappings) if not delete_failed: @@ -1156,13 +1345,17 @@ def sync_unified_state(): k210_base_timestamp = 0 # 基准时间戳（Unix时间戳） k210_base_ticks = 0 # 基准时刻的ticks +# 函数作用：根据服务端时间设置 K210 本地时间基准。 def set_base_time(unix_timestamp): """设置K210的基准时间""" + # K210 不带可靠 RTC，这里用“服务端/STM32 给定的 Unix 时间 + 本机 ticks 增量” + # 估算当前时间，适合生成临时日志时间或调试信息。 global k210_base_timestamp, k210_base_ticks k210_base_timestamp = unix_timestamp k210_base_ticks = utime.ticks_ms() print("[TIME] Base time set:", unix_timestamp) +# 函数作用：基于本地时间基准生成当前时间戳字符串。 def get_current_timestamp(): """获取当前Unix时间戳""" if k210_base_timestamp == 0: @@ -1174,6 +1367,10 @@ def get_current_timestamp(): return k210_base_timestamp + elapsed_seconds # --- 2. KPU模型和硬件初始化 --- +# 三个模型分别对应人脸识别流水线的三个阶段： +# task_fd：FaceDetection，检测画面中是否有人脸以及人脸框位置； +# task_ld：FaceLandmarkDetection，定位眼睛、鼻子、嘴角等关键点； +# task_fe：FeatureExtraction，提取 128 维左右的人脸特征向量用于比对。 try: task_fd = kpu.load("/sd/FaceDetection.smodel") task_ld = kpu.load("/sd/FaceLandmarkDetection.smodel") @@ -1184,6 +1381,11 @@ except Exception as e: utime.sleep_ms(1000) clock = time.clock() + +# RGB LED 引脚映射。这里高电平为灭、低电平为亮： +# 绿灯亮：识别到已知人员； +# 红灯亮：检测到陌生人、删除警告或操作失败； +# 全灭/全亮状态用于提示不同运行阶段。 fm.register(12, fm.fpioa.GPIO0) fm.register(13, fm.fpioa.GPIO1) fm.register(14, fm.fpioa.GPIO2) @@ -1193,6 +1395,9 @@ LED_R = GPIO(GPIO.GPIO2, GPIO.OUT, value=1) key_pin = 16 fm.register(key_pin, fm.fpioa.GPIOHS0) KEY = GPIO(GPIO.GPIOHS0, GPIO.IN) + +# 摄像头和 LCD 初始化。 +# QVGA(320x240) 是 K210 人脸检测常用分辨率，速度和识别效果比较平衡。 lcd.init() sensor.reset() sensor.set_pixformat(sensor.RGB565) @@ -1202,7 +1407,15 @@ sensor.set_vflip(CAMERA_VFLIP) sensor.run(1) anchor = (1.889, 2.5245, 2.9465, 3.94056, 3.99987, 5.3658, 5.155437, 6.92275, 6.718375, 9.01025) dst_point = [(44, 59), (84, 59), (64, 82), (47, 105), (81, 105)] + +# 初始化 YOLOv2 人脸检测参数： +# 0.5 为置信度阈值，低于该分数的人脸框会被过滤； +# 0.3 为 NMS 阈值，用于去掉重叠框； +# anchor 是模型训练时使用的人脸框先验尺寸。 kpu.init_yolo2(task_fd, 0.5, 0.3, 5, anchor) + +# img_face 是送入特征提取模型的标准 128x128 人脸图。 +# 检测到人脸后，会根据关键点把原图人脸仿射对齐到这个画布上。 img_face = image.Image(size=(128, 128)) img_face.pix_to_ai() @@ -1210,7 +1423,10 @@ img_face.pix_to_ai() key_Flag = 0 key_pressed = 0 # 0=无, 1=短按, 2=长按(2秒), 3=超长按(5秒删除全部) t_key = 0 +# 函数作用：处理按键中断并切换人脸录入相关状态。 def fun(KEY): + # 按键中断只负责记录“按下持续时间”对应的事件类型，不在中断里做耗时操作。 + # 主循环随后根据 key_pressed 切换到添加、删除或清空模式。 global key_Flag, key_pressed, t_key utime.sleep_ms(10) if KEY.value() == 0: @@ -1229,16 +1445,17 @@ def fun(KEY): key_Flag = 0 KEY.irq(fun, GPIO.IRQ_BOTH) +# 函数作用：清空本地人脸特征库和相关照片缓存。 def delete_all_faces(): """删除全部人脸数据""" global peoples, face_id_mapping try: - # 清空人脸列表 + # 清空内存中的人脸列表。 peoples = [] - # 清空记录文件 + # 清空 SD 卡特征记录文件。 with open(record_path, "w") as f: pass # 写入空文件 - # 清空映射 + # 清空学号映射和本地照片，避免服务端或下次启动恢复旧状态。 face_id_mapping = {} save_local_mapping() delete_all_local_face_photos() @@ -1249,16 +1466,26 @@ def delete_all_faces(): load_faces() return False -# --- 4. 数据库和人脸管理函数 (已包含之前修复的bug) --- +# --- 4. 数据库和人脸管理函数 --- +# recordftr.txt 每行保存一张人脸： +# ID:1#bytearray(b"...") +# 左侧是 K210 原始 face_id，右侧是 KPU 提取的人脸特征。 +# 由于历史版本可能把特征保存成 bytearray 文本、bytes 文本或数字列表， +# 下面的解析函数专门做兼容读取，避免使用 eval 带来的安全和稳定性问题。 class People: + # 函数作用：初始化人脸人员对象，保存名称和特征向量。 def __init__(self, name="", feature=[]): self.name = name; self.feature = feature + # 函数作用：计算输入特征与当前人员特征集合的最大相似度。 def feature_max(self, feature): + # 返回当前人脸特征与待比对特征的相似度分数，分数越高表示越像。 if not self.feature: return 0 return kpu.face_compare(self.feature, feature) peoples = [] record_path = "/sd/recordftr.txt" +# 函数作用：把单个十六进制字符转换为整数。 def hex_value(ch): + # 将单个十六进制字符转换为 0~15，用于解析 \xAB 形式的字节转义。 if '0' <= ch <= '9': return ord(ch) - ord('0') if 'a' <= ch <= 'f': @@ -1267,6 +1494,7 @@ def hex_value(ch): return ord(ch) - ord('A') + 10 return -1 +# 函数作用：解析 Python bytes 字面量文本为字节值列表。 def parse_bytes_literal(text): """安全解析 bytearray(b"...") / b"..." 文本，不使用 eval。""" pos1 = text.find("b'") @@ -1285,14 +1513,17 @@ def parse_bytes_literal(text): while i < len(text): ch = text[i] if ch == quote: + # 特征向量长度过短通常说明文件损坏或截断，直接拒绝加载该行。 if len(values) < 64: raise ValueError("feature too short") return bytearray(values) if ch != "\\": + # 普通可见字符按 ASCII/字节值写入。 values.append(ord(ch)) i += 1 else: + # 处理 Python bytes 文本中的转义序列，例如 \x12、\n、\\ 等。 i += 1 if i >= len(text): raise ValueError("bad escape") @@ -1347,11 +1578,15 @@ def parse_bytes_literal(text): i += 1 if len(values) > 4096: + # 正常人脸特征远小于 4096 字节，超过该值基本可以判断为异常数据。 raise ValueError("feature too long") raise ValueError("unterminated bytes literal") +# 函数作用：解析逗号分隔的数字列表文本。 def parse_number_list(text): + # 兼容 [1, 2, 3] 或 [0.1, 0.2, ...] 形式的历史特征记录。 + # 若全部数值都是 0~255 的整数，则转为 bytearray；否则保留浮点列表。 start = text.find("[") end = text.rfind("]") if start < 0 or end <= start: @@ -1386,7 +1621,9 @@ def parse_number_list(text): return bytearray(values) return values +# 函数作用：解析人脸特征向量的文本存储格式。 def parse_feature_vector(text): + # 根据文本前缀自动选择解析方式，统一返回 kpu.face_compare 可接受的特征对象。 text = text.strip() if text.startswith("bytearray(") or text.startswith("b'") or text.startswith('b"'): return parse_bytes_literal(text) @@ -1394,10 +1631,15 @@ def parse_feature_vector(text): HEX_CHARS = "0123456789abcdef" +# 函数作用：把单个字节值转换为两位十六进制文本。 def byte_to_hex(v): + # 不依赖格式化函数生成两位十六进制，减少 MaixPy 兼容性问题。 return HEX_CHARS[(v >> 4) & 15] + HEX_CHARS[v & 15] +# 函数作用：把字节值列表转换为可写入记录文件的转义文本。 def escape_byte_values(values): + # 将 bytearray 转回可读的 bytes 字面量文本。 + # 可见字符直接写，控制字符和非 ASCII 字节使用转义，便于下次安全解析。 out = "" for v in values: v = int(v) & 0xff @@ -1417,7 +1659,10 @@ def escape_byte_values(values): out += "\\x" + byte_to_hex(v) return out +# 函数作用：把人脸特征向量转换为本地记录文件文本。 def feature_to_record_text(feature): + # 保存特征时尽量保持 bytearray(b"...") 这种紧凑格式； + # 若当前特征对象不是 bytes/bytearray，则退回 str(feature)。 type_text = str(type(feature)) if type_text.find("bytearray") >= 0 or type_text.find("bytes") >= 0: values = [] @@ -1426,12 +1671,18 @@ def feature_to_record_text(feature): return 'bytearray(b"' + escape_byte_values(values) + '")' return str(feature) +# 函数作用：把当前内存中的人脸库写入 SD 卡记录文件。 def write_face_record(): + # 将内存中的 peoples 完整写回 recordftr.txt。 + # 删除人脸时采用重写文件的方式，保证文件里不残留旧记录。 with open(record_path, "w") as f: for p in peoples: f.write(str(p.name) + '#' + feature_to_record_text(p.feature) + "\n") +# 函数作用：从 SD 卡记录文件加载本地人脸特征库。 def load_faces(): + # 从 SD 卡加载本地人脸特征库。 + # 单行解析失败不会终止整个加载过程，只统计失败数量并继续读取后续记录。 global peoples; peoples = [] total_count = 0 bad_count = 0 @@ -1459,20 +1710,24 @@ def load_faces(): print("成功加载 %d 条人脸数据 (总行:%d, 失败:%d)" % (len(peoples), total_count, bad_count)) except Exception as e: print("读取人脸数据库失败:", e) +# 函数作用：收集当前本地人脸库中已经使用的人脸编号。 def collect_used_face_ids(): """收集本地已经占用的K210原始face_id，避免新增时覆盖旧人脸。""" used = {} + # 先从特征库收集已存在的 ID。 for p in peoples: name = str(p.name) if name.startswith("ID:") and name.split(":", 1)[1].isdigit(): used[name.split(":", 1)[1]] = 1 + # 再从映射文件收集 ID，防止特征和映射短暂不同步时重复分配。 for face_id in face_id_mapping.keys(): face_id = str(face_id) if face_id.isdigit(): used[face_id] = 1 + # 最后从本地照片目录收集 ID，避免存在照片但特征尚未成功写入时被覆盖。 if ensure_local_photo_dir(): try: for filename in os.listdir(LOCAL_FACE_PHOTO_DIR): @@ -1485,7 +1740,9 @@ def collect_used_face_ids(): return used +# 函数作用：分配下一个可用的人脸编号。 def allocate_next_face_id(): + # 从 1 开始寻找最小可用 face_id，便于现场查看和管理端展示。 used = collect_used_face_ids() next_id = 1 while str(next_id) in used: @@ -1493,7 +1750,13 @@ def allocate_next_face_id(): print("[ADD] 已占用face_id: {}, 新face_id: {}".format(sorted(used.keys()), next_id)) return next_id +# 函数作用：新增人脸特征记录，并保存或上传对应照片。 def add_face(feature, face_img=None): + # 新增人脸的顺序很重要： + # 1. 分配未占用 face_id； + # 2. 先保存照片并加入上传队列； + # 3. 再把特征追加写入 recordftr.txt。 + # 若任一步失败，会回滚内存列表和照片，避免出现“有特征无照片”的半完成状态。 new_id_num = allocate_next_face_id() new_name = "ID:{}".format(new_id_num) peoples.append(People(new_name, feature)) @@ -1515,7 +1778,10 @@ def add_face(feature, face_img=None): delete_local_face_photo(new_id_num) peoples.pop() return False +# 函数作用：按本地列表索引删除人脸记录。 def delete_face_by_index(idx): + # 本地按索引删除，用于 K210 屏幕上的手动删除模式。 + # 删除后重写特征文件，并同步删除对应照片。 if not (0 <= idx < len(peoples)): return False deleted_name = peoples[idx].name peoples.pop(idx) @@ -1528,8 +1794,15 @@ def delete_face_by_index(idx): print("删除 %s 成功" % deleted_name); return True except Exception as e: print("重写文件失败:", e); load_faces(); return False -# --- 5. 支持多张人脸检测的函数 (保持不变) --- +# --- 5. 支持多张人脸检测的函数 --- +# 函数作用：检测图像中的人脸并执行本地特征识别。 def detect_and_recognize_faces(img): + # 完整识别流程： + # 1. YOLOv2 检测当前画面中的全部人脸框； + # 2. 对每张脸做人脸关键点检测； + # 3. 根据关键点做仿射变换，将人脸对齐到 128x128 标准图； + # 4. 提取特征并与本地特征库逐一比对； + # 5. 分数超过阈值认为是已知人员，否则标记为 stranger。 codes = kpu.run_yolo2(task_fd, img) if not codes: return [] results = [] @@ -1538,18 +1811,22 @@ def detect_and_recognize_faces(img): face_cut_128 = face_cut.resize(128, 128); face_cut_128.pix_to_ai() fmap = kpu.forward(task_ld, face_cut_128) plist = fmap[:] + # 根据模型输出的五点关键点计算左眼、右眼、鼻尖、左嘴角、右嘴角坐标。 le=(item.x()+int(plist[0]*item.w()-10), item.y()+int(plist[1]*item.h())); re=(item.x()+int(plist[2]*item.w()), item.y()+int(plist[3]*item.h())); nose=(item.x()+int(plist[4]*item.w()), item.y()+int(plist[5]*item.h())); lm=(item.x()+int(plist[6]*item.w()), item.y()+int(plist[7]*item.h())); rm=(item.x()+int(plist[8]*item.w()), item.y()+int(plist[9]*item.h())) src_point = [le, re, nose, lm, rm] + # 将当前人脸的五点关键点映射到固定模板点 dst_point，减少姿态和尺度差异。 T = image.get_affine_transform(src_point, dst_point) image.warp_affine_ai(img, img_face, T) del(face_cut); del(face_cut_128) fmap = kpu.forward(task_fe, img_face) feature = kpu.face_encode(fmap[:]) max_score = 0; match_index = -1 + # 与本地所有人脸逐一比较，取相似度最高的一条作为识别结果。 for i, p in enumerate(peoples): score = p.feature_max(feature) if score > max_score: max_score = score; match_index = i result_item = {'score': max_score, 'rect': item.rect(), 'feature': feature} + # 85 是经验阈值：过低容易误识别，过高会降低识别通过率。 if max_score > 85: result_item['state'] = peoples[match_index].name else: result_item['state'] = "stranger" results.append(result_item) @@ -1558,11 +1835,12 @@ def detect_and_recognize_faces(img): # --- 6. 主程序启动前准备 - 添加发送逻辑所需变量 --- load_faces() last_send_time = 0 -SEND_INTERVAL_MS = 1000 -last_sent_data = "" # 改为记录完整数据而不是state -last_saved_id = "" # ✨ 添加：初始化上次保存的学号 +SEND_INTERVAL_MS = 1000 # 串口上报最小间隔，避免同一张脸每帧都发送给 STM32 +last_sent_data = "" # 记录上一次完整串口消息，内容不变时不重复发送 +last_saved_id = "" # 预留字段：记录最近一次保存/处理的学号，切换状态时重置 # --- 分时网络：照片上传队列操作 --- +# 函数作用：把指定人脸编号加入照片待上传队列。 def queue_photo_upload(face_id): """将本地照片ID添加到上传队列（非阻塞）""" global photo_upload_queue @@ -1573,8 +1851,11 @@ def queue_photo_upload(face_id): photo_upload_queue.append(face_id) print("[QUEUE] 照片已加入队列: ID={}，队列长度: {}".format(face_id, len(photo_upload_queue))) +# 函数作用：扫描本地照片缓存并加入待上传队列。 def queue_local_photos_for_upload(): """启动时扫描SD卡照片，补传可能遗漏到服务器的照片""" + # K210 可能在离线状态下新增人脸，此时照片只能先存到 SD 卡。 + # 启动时先向服务端查询已有照片，服务端没有的才加入队列，减少重复上传。 if not ensure_local_photo_dir(): return @@ -1606,11 +1887,13 @@ def queue_local_photos_for_upload(): except Exception as e: print("[QUEUE] 扫描本地照片失败:", e) +# 函数作用：执行周期性网络任务，包括上传照片和同步服务端状态。 def do_network_task(): """执行一个网络任务（每次只做一件事）""" global photo_upload_queue, last_sync_time - # 优先处理照片上传 + # 当前分时任务优先处理照片上传。 + # 上传失败会把 face_id 放回队尾，等待下次调度继续重试。 if len(photo_upload_queue) > 0: face_id = photo_upload_queue.pop(0) jpeg_data = load_local_face_photo(face_id) @@ -1647,9 +1930,10 @@ else: print("[BOOT] WiFi未连接，跳过同步") # QT触发同步标志 -sync_requested = False -last_sync_check_time = 0 +sync_requested = False # 服务端提示需要同步后置为 True，主循环空闲时执行 trigger_sync +last_sync_check_time = 0 # 上一次检查服务端同步请求的时间 +# 函数作用：检查服务端是否要求 K210 立即同步。 def check_sync_request(): """检查服务器是否有同步请求""" global sync_requested @@ -1661,13 +1945,14 @@ def check_sync_request(): if response and '"sync":true' in response: print("[SYNC] 收到服务器同步请求") sync_requested = True - # 确认收到请求 + # 确认收到请求。实际同步放到主循环中执行，避免接口轮询阶段连续阻塞。 esp_http_post_json("/api/v1/sync/request/ack", '{"ack":true}', 1000) return True except Exception as e: print("[SYNC] 检查同步请求错误:", e) return False +# 函数作用：主动触发一次服务端统一状态同步。 def trigger_sync(): """执行一次完整同步（由QT端触发）""" global sync_requested @@ -1686,15 +1971,23 @@ def trigger_sync(): return True # --- 7. 主循环 (状态机) --- +# 主循环按 key_pressed 状态分为四种模式： +# 0：默认识别模式，持续采集画面、识别人脸，并按需发送给 STM32； +# 1：添加人脸模式，短按确认把当前陌生人保存为新 face_id； +# 2：删除人脸模式，短按切换条目，长按确认删除； +# 3：删除全部人脸模式，超长按进入二次确认。 +# 每轮循环末尾还会做一次轻量的网络调度和 gc.collect()，保证识别任务长期稳定运行。 while True: clock.tick() # ------------------ 模式0：人脸识别与智能通信 (多人脸 & 无脸不发送) ------------------ if key_pressed == 0: + # 默认模式下，每帧采集图像并识别当前画面中的全部人脸。 img = sensor.snapshot() detected_faces = detect_and_recognize_faces(img) current_state = "" recognized_ids = [] if not detected_faces: + # 无人脸时关闭红绿灯，不主动给 STM32 发送“无人脸”，减少串口噪声。 LED_G.value(1); LED_R.value(1) else: has_stranger = False; has_known_person = False @@ -1702,30 +1995,35 @@ while True: rect = face['rect'] img.draw_rectangle(rect) if face['state'] == "stranger": + # 陌生人用红色标记，同时在待发送状态中记为 0。 img.draw_string(rect[0], rect[1], "Stranger", color=(255, 0, 0), scale=2) recognized_ids.append("0"); has_stranger = True else: - # 使用映射后的学号显示 + # 已知人脸优先显示映射后的学号；未绑定时显示原始 ID:x。 display_name = get_display_name(face['state']) img.draw_string(rect[0], rect[1], display_name, color=(0, 255, 0), scale=2) recognized_ids.append(face['state']); has_known_person = True + # LED 状态提示识别结果：绿灯代表已知人员，红灯代表只检测到陌生人。 if has_known_person: LED_G.value(0); LED_R.value(1) elif has_stranger: LED_G.value(1); LED_R.value(0) recognized_ids.sort(); current_state = ",".join(recognized_ids) time_now = utime.ticks_ms() if utime.ticks_diff(time_now, last_send_time) > SEND_INTERVAL_MS: if detected_faces: - # 发送第一个识别到的人脸数据（可以改为发送得分最高的） + # 多人同时出现时，选择相似度最高的人脸发送给 STM32。 + # 这样比“第一张脸”更稳定，能减少画面中小人脸或边缘人脸误触发。 best_face = max(detected_faces, key=lambda x: x['score']) if best_face['state'] == "stranger": - # 陌生人，发送UNKNOWN + # 陌生人发送 UNKNOWN，STM32 可据此提示未注册或拒绝签到。 msg_to_send = "UNKNOWN\r\n" - # 陌生人不保存到SD卡 + # 默认识别模式下陌生人不自动入库，只有进入添加模式后才保存。 last_saved_id = "" # 重置保存记录 else: - # 已识别用户，发送完整信息（STM32期望的格式） + # 已识别用户发送 STM32 约定格式： + # ID:<学号或face_id>,NAME:<显示名>,CONF:<置信度> + # 若管理端尚未绑定学号，则使用 K210 原始 face_id 兜底，保证系统仍能跑通。 face_id = extract_face_id(best_face['state']) mapped_student_id = get_mapped_student_id(best_face['state']) if not mapped_student_id: @@ -1734,7 +2032,7 @@ while True: student_name = mapped_student_id if mapped_student_id else "FACE_{}".format(face_id) confidence = best_face['score'] / 100.0 - # 构造STM32期望的格式；未绑定学号时默认使用K210原始face_id。 + # 构造 STM32 期望的格式；未绑定学号时默认使用 K210 原始 face_id。 msg_to_send = "ID:{},NAME:{},CONF:{:.2f}\r\n".format( student_id, student_name, @@ -1742,12 +2040,13 @@ while True: ) if msg_to_send != last_sent_data: + # 只有识别结果变化时才发送，避免 STM32 重复登记同一人。 uart.write(msg_to_send) print("[UART SEND] -> {}".format(msg_to_send.strip())) last_sent_data = msg_to_send last_send_time = time_now else: - # 没有检测到人脸 + # 从“有人脸”变为“无人脸”时只重置本地状态，不发送额外消息。 if last_sent_data != "": print("[STATE CHANGE] Faces disappeared. No data sent.") last_sent_data = "" @@ -1755,9 +2054,10 @@ while True: last_send_time = time_now lcd.display(img) - # ------------------ 【核心修改】模式1：添加人脸 (添加了实时框) ------------------ + # ------------------ 模式1：添加人脸 ------------------ elif key_pressed == 1: key_pressed = 0 + # 进入管理模式前清空上报状态，退出后下一次识别会重新发送。 last_sent_data = "" # 修改变量名 last_saved_id = "" # 重置保存记录 last_send_time = 0 @@ -1769,27 +2069,28 @@ while True: detected_faces = detect_and_recognize_faces(img) best_face = None if detected_faces: - # 选出最清晰（得分最高）的脸进行操作 + # 选出相似度最高/质量最稳定的脸进行操作。 best_face = max(detected_faces, key=lambda x: x['score']) if best_face: rect = best_face['rect'] # 获取目标人脸的框 if best_face['state'] != "stranger": - # 【新】如果是已存在的用户，画一个橙色的框 + # 已存在的人脸不能重复添加，橙色框提示用户换人或退出。 img.draw_rectangle(rect, color=(255, 128, 0), thickness=2) img.draw_string(rect[0], rect[1], "Exists!", color=(255, 128, 0), scale=2) else: - # 【新】如果是可以添加的陌生人，画一个绿色的框 + # 陌生人可以作为新 face_id 添加，绿色框表示当前画面可确认。 img.draw_rectangle(rect, color=(0, 255, 0), thickness=2) img.draw_string(rect[0], rect[1], "Ready", color=(0, 255, 0), scale=2) - # 按键确认逻辑 (保持不变) + # 按键确认逻辑：短按确认添加，长按退出添加模式。 if key_pressed == 1: key_pressed = 0 if best_face: if best_face['state'] != "stranger": img.draw_string(best_face['rect'][0], best_face['rect'][1] + 24, "Already exists!", color=(255, 0, 0), scale=2); LED_R.value(0) else: + # 新增时保存特征，同时把当前画面压缩为照片并加入上传队列。 if add_face(best_face['feature'], img): img.draw_string(best_face['rect'][0], best_face['rect'][1] + 24, "Add OK!", color=(0, 255, 0), scale=2); LED_G.value(0) else: @@ -1802,7 +2103,7 @@ while True: lcd.display(img) print("退出添加人脸模式") - # ------------------ 模式2：删除人脸 (保持不变) ------------------ + # ------------------ 模式2：删除人脸 ------------------ elif key_pressed == 2: key_pressed = 0 last_sent_data = "" # 修改变量名 @@ -1814,11 +2115,13 @@ while True: img = image.Image(); img.clear() img.draw_string(10, 10, "Delete Mode", color=(255, 0, 0), scale=2); img.draw_string(10, 40, "Click=Next, Long Press=Confirm", scale=1.8) if key_pressed == 1: + # 短按在“取消”和各个人脸条目之间循环选择。 key_pressed = 0; select_index += 1 if select_index >= len(peoples): select_index = -1 if select_index == -1: img.draw_string(50, 100, "-> Cancel", color=(0, 255, 0), scale=2) else: img.draw_string(50, 100, "-> Del: {}".format(peoples[select_index].name), color=(255, 128, 0), scale=2) if key_pressed == 2: + # 长按确认当前选择；select_index=-1 表示取消删除。 key_pressed = 0 if select_index != -1: if delete_face_by_index(select_index): @@ -1843,7 +2146,7 @@ while True: lcd.display(img) LED_R.value(0) # 红灯警告 - # 等待用户确认 + # 删除全部人脸属于高风险操作，因此进入该模式后还需要二次确认。 confirm_timeout = utime.ticks_ms() while utime.ticks_diff(utime.ticks_ms(), confirm_timeout) < 10000: # 10秒超时 if key_pressed == 1: # 短按确认删除 @@ -1874,6 +2177,11 @@ while True: print("退出删除全部人脸模式") # 分时网络处理：每N帧检查一次网络任务（只处理照片上传和同步请求） + # 网络调度放在所有模式处理之后执行。 + # 每次只做一类任务，优先级为： + # 1. 上传本地照片； + # 2. 执行已收到的同步请求； + # 3. 定时向服务端询问是否需要同步。 frames_since_network += 1 if frames_since_network >= NETWORK_FRAME_INTERVAL: frames_since_network = 0