2201 lines
92 KiB
Python
2201 lines
92 KiB
Python
# ====================================================================================
|
||
# K210 V11 (分时网络通信)
|
||
# 本文件运行在 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
|
||
import KPU as kpu
|
||
import time
|
||
from Maix import FPIOA, GPIO, utils
|
||
import gc
|
||
from fpioa_manager import fm
|
||
import utime
|
||
from machine import UART
|
||
import os
|
||
try:
|
||
import ubinascii
|
||
except:
|
||
ubinascii = None
|
||
try:
|
||
import ucryptolib as crypto_lib
|
||
except:
|
||
try:
|
||
import cryptolib as crypto_lib
|
||
except:
|
||
crypto_lib = None
|
||
|
||
# --- 增加堆内存大小 ---
|
||
# 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 帧检查一次网络任务；数值越小同步越及时，但识别画面越容易卡顿
|
||
|
||
# --- 分时网络任务队列 ---
|
||
photo_upload_queue = [] # 照片上传队列，保存待上传的 face_id；真正的 JPEG 数据存放在 SD 卡
|
||
network_task_pending = False # 预留的网络任务状态位，目前主要由照片队列和同步标志驱动
|
||
frames_since_network = 0 # 距离上次网络操作已经经过的帧数，用于在主循环中做分时调度
|
||
|
||
# --- 0. ESP8266 WiFi配置 ---
|
||
# 这些参数需要与实际部署环境保持一致：
|
||
# - 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]
|
||
if pad_len < 1 or pad_len > 16 or pad_len > len(data):
|
||
return data
|
||
for b in data[-pad_len:]:
|
||
if b != pad_len:
|
||
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
|
||
try:
|
||
cipher = crypto_lib.aes(AES_KEY, 2, AES_IV)
|
||
return cipher.encrypt(data) if encrypt else cipher.decrypt(data)
|
||
except Exception as e:
|
||
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)
|
||
if encrypted is None:
|
||
return None
|
||
encoded = ubinascii.b2a_base64(encrypted).strip()
|
||
try:
|
||
return encoded.decode("utf-8")
|
||
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):
|
||
data = data.strip().encode("utf-8")
|
||
try:
|
||
encrypted = ubinascii.a2b_base64(data)
|
||
except Exception as e:
|
||
print("[CRYPTO] Base64 decode failed:", e)
|
||
return None
|
||
decrypted = aes_cbc_crypt(encrypted, False)
|
||
if decrypted is None:
|
||
return None
|
||
try:
|
||
return pkcs7_unpad(decrypted).decode("utf-8")
|
||
except Exception as e:
|
||
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"
|
||
start = resp.find(marker)
|
||
if start == -1:
|
||
return ""
|
||
body = resp[start + len(marker):]
|
||
content_length = -1
|
||
cl_pos = resp.find("Content-Length:")
|
||
if cl_pos != -1:
|
||
cl_pos += len("Content-Length:")
|
||
while cl_pos < len(resp) and resp[cl_pos] == " ":
|
||
cl_pos += 1
|
||
cl_end = cl_pos
|
||
while cl_end < len(resp) and resp[cl_end].isdigit():
|
||
cl_end += 1
|
||
try:
|
||
content_length = int(resp[cl_pos:cl_end])
|
||
except:
|
||
content_length = -1
|
||
if content_length >= 0 and len(body) >= content_length:
|
||
return body[:content_length]
|
||
for marker in ("\r\nCLOSED", "CLOSED", "\r\nOK"):
|
||
pos = body.find(marker)
|
||
if pos != -1:
|
||
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]
|
||
cl_pos = resp.find("Content-Length:")
|
||
if cl_pos == -1:
|
||
return "CLOSED" in resp
|
||
cl_pos += len("Content-Length:")
|
||
while cl_pos < len(resp) and resp[cl_pos] == " ":
|
||
cl_pos += 1
|
||
cl_end = cl_pos
|
||
while cl_end < len(resp) and resp[cl_end].isdigit():
|
||
cl_end += 1
|
||
try:
|
||
return len(body) >= int(resp[cl_pos:cl_end])
|
||
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
|
||
if "X-Encrypted: AES128-CBC" in resp:
|
||
return decrypt_payload_base64(body)
|
||
if body.startswith("{") or body.startswith("["):
|
||
return body
|
||
return decrypt_payload_base64(body)
|
||
|
||
# 摄像头安装方向：当前实物画面上下左右都反，水平镜像+垂直翻转等价于旋转180度。
|
||
CAMERA_HMIRROR = 1
|
||
CAMERA_VFLIP = 1
|
||
|
||
# --- 1. UART通信初始化 ---
|
||
# 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。
|
||
# 该串口只负责网络通信：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
|
||
ESP_UART_BAUD = baud
|
||
try:
|
||
uart2.deinit()
|
||
except:
|
||
pass
|
||
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()
|
||
uart2.write(cmd + "\r\n")
|
||
utime.sleep_ms(50)
|
||
start = utime.ticks_ms()
|
||
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()
|
||
if data:
|
||
response += data.decode('utf-8', 'ignore')
|
||
if any(token in response for token in expected):
|
||
break
|
||
utime.sleep_ms(5)
|
||
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)
|
||
esp_reopen_uart(baud)
|
||
for _ in range(2):
|
||
resp = esp_send_cmd("AT", 1500)
|
||
if "OK" in resp:
|
||
print("[ESP] UART baud detected:", baud)
|
||
return True
|
||
utime.sleep_ms(300)
|
||
esp_reopen_uart(115200)
|
||
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...")
|
||
start = utime.ticks_ms()
|
||
boot_log = ""
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < ESP_BOOT_WAIT_MS:
|
||
if uart2.any():
|
||
data = uart2.read()
|
||
if data:
|
||
boot_log += data.decode('utf-8', 'ignore')
|
||
utime.sleep_ms(100)
|
||
if ESP_DEBUG and boot_log:
|
||
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) # 关闭命令回显，后续解析响应更干净
|
||
|
||
# 使用 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)
|
||
|
||
if "OK" not in resp and "no change" not in resp:
|
||
print("[ESP] Set mode failed")
|
||
return False
|
||
|
||
if esp_check_wifi_status():
|
||
print("[ESP] WiFi already connected!")
|
||
return True
|
||
|
||
# 连接 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"))
|
||
|
||
if "OK" not in resp and "CONNECTED" not in resp and "WIFI GOT IP" not in resp:
|
||
cmd = 'AT+CWJAP="{}","{}"'.format(WIFI_SSID, WIFI_PASSWORD)
|
||
resp = esp_send_cmd(cmd, 20000, ("WIFI GOT IP", "OK", "FAIL", "ERROR"))
|
||
|
||
if "OK" in resp or "CONNECTED" in resp or "WIFI GOT IP" in resp:
|
||
utime.sleep_ms(1000)
|
||
status = esp_send_cmd("AT+CWJAP?", 1500)
|
||
ip = esp_send_cmd("AT+CIFSR", 1500)
|
||
if WIFI_SSID not in status and "STAIP" not in ip:
|
||
print("[ESP] WiFi command returned OK, but no station IP found")
|
||
return False
|
||
print("[ESP] WiFi connected!")
|
||
return True
|
||
|
||
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 响应。若默认波特率没有响应，再调用 esp_probe_uart 自动探测。
|
||
for i in range(max_retries):
|
||
resp = esp_send_cmd("AT", 2500)
|
||
if "OK" in resp:
|
||
print("[ESP] AT OK")
|
||
break
|
||
print("[ESP] AT retry {}/{}".format(i+1, max_retries))
|
||
utime.sleep_ms(500)
|
||
else:
|
||
if not esp_probe_uart():
|
||
print("[ESP] ESP8266 not responding!")
|
||
return False
|
||
|
||
# 连接 WiFi。初始化阶段允许多次重试，运行中重连时通常只传入 1 次，避免阻塞太久。
|
||
for i in range(max_retries):
|
||
if esp_connect_wifi():
|
||
return True
|
||
utime.sleep_ms(1000)
|
||
|
||
return False
|
||
|
||
# 函数作用：确保 WiFi 连接可用，断开时自动重连。
|
||
def esp_ensure_wifi():
|
||
"""确保WiFi已连接（快速检查）"""
|
||
global wifi_connected
|
||
if not wifi_connected:
|
||
wifi_connected = esp_init_wifi(1)
|
||
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
|
||
except:
|
||
try:
|
||
os.mkdir(LOCAL_FACE_PHOTO_DIR)
|
||
print("[PHOTO] Created local dir:", LOCAL_FACE_PHOTO_DIR)
|
||
return True
|
||
except Exception as e:
|
||
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)
|
||
try:
|
||
with open(path, "wb") as f:
|
||
f.write(jpeg_data)
|
||
print("[PHOTO] Saved local photo:", path)
|
||
return True
|
||
except Exception as e:
|
||
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:
|
||
return f.read()
|
||
except Exception as e:
|
||
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)
|
||
try:
|
||
os.remove(path)
|
||
print("[PHOTO] Deleted local photo:", path)
|
||
return True
|
||
except:
|
||
return False
|
||
|
||
# 函数作用：删除所有本地人脸照片缓存。
|
||
def delete_all_local_face_photos():
|
||
# 清空全部人脸时同步清理照片目录和上传队列，保持 SD 卡状态一致。
|
||
if not ensure_local_photo_dir():
|
||
return
|
||
try:
|
||
for name in os.listdir(LOCAL_FACE_PHOTO_DIR):
|
||
if name.endswith(".jpg"):
|
||
try:
|
||
os.remove("{}/{}".format(LOCAL_FACE_PHOTO_DIR, name))
|
||
except:
|
||
pass
|
||
photo_upload_queue[:] = []
|
||
print("[PHOTO] Deleted 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
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < 1000:
|
||
if uart2.any():
|
||
resp += uart2.read().decode('utf-8', 'ignore')
|
||
if ">" in resp:
|
||
got_prompt = True
|
||
break
|
||
if "ERROR" in resp or "link is not" in resp:
|
||
return False
|
||
utime.sleep_ms(5)
|
||
|
||
if not got_prompt:
|
||
return False
|
||
|
||
# 发送实际数据。data 既可能是字符串，也可能是 JPEG bytes。
|
||
uart2.write(data)
|
||
|
||
# 等待 SEND OK。超时设置偏短，是为了网络异常时不长时间卡住识别。
|
||
start = utime.ticks_ms()
|
||
resp = ""
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < timeout:
|
||
if uart2.any():
|
||
resp += uart2.read().decode('utf-8', 'ignore')
|
||
if "SEND OK" in resp:
|
||
return True
|
||
if "SEND FAIL" in resp or "ERROR" in resp:
|
||
return False
|
||
utime.sleep_ms(10)
|
||
|
||
return False
|
||
|
||
# 函数作用：通过 HTTP POST 上传指定人脸照片到服务端。
|
||
def esp_http_post_photo(face_id, jpeg_data, max_retries=2):
|
||
"""通过ESP8266发送HTTP POST请求上传照片"""
|
||
# 直接使用 face_id 作为照片 ID（不加时间戳），服务端可用该 ID 覆盖或更新同一张人脸照片。
|
||
print("[ESP] Uploading photo for face ID:", face_id, "size:", len(jpeg_data))
|
||
|
||
for attempt in range(max_retries):
|
||
# 上传前确认 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)
|
||
|
||
# 建立到服务端 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:
|
||
print("[ESP] TCP failed, retry {}/{}".format(attempt+1, max_retries))
|
||
utime.sleep_ms(1000)
|
||
continue
|
||
|
||
print("[ESP] TCP connected, sending data...")
|
||
|
||
# 构造照片上传请求。
|
||
# 这里 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"
|
||
http_header += "X-Api-Token: {}\r\n".format(API_TOKEN)
|
||
http_header += "Content-Length: {}\r\n".format(len(jpeg_data))
|
||
http_header += "Connection: close\r\n\r\n"
|
||
|
||
# 先发送 HTTP 头。照片 body 后续分块发送。
|
||
if not esp_send_data(http_header):
|
||
print("[ESP] Send header failed")
|
||
esp_send_cmd("AT+CIPCLOSE", 500)
|
||
continue
|
||
|
||
# 分块发送 JPEG 数据。
|
||
# ESP8266 AT 透传缓冲区有限，大照片一次性发送容易失败，按 1024 字节拆分更稳。
|
||
CHUNK_SIZE = 1024
|
||
offset = 0
|
||
send_ok = True
|
||
|
||
while offset < len(jpeg_data):
|
||
chunk = jpeg_data[offset:offset + CHUNK_SIZE]
|
||
if not esp_send_data(chunk):
|
||
print("[ESP] Send chunk failed at", offset)
|
||
send_ok = False
|
||
break
|
||
offset += CHUNK_SIZE
|
||
print("[ESP] Sent {}/{}".format(min(offset, len(jpeg_data)), len(jpeg_data)))
|
||
|
||
if not send_ok:
|
||
esp_send_cmd("AT+CIPCLOSE", 500)
|
||
utime.sleep_ms(500)
|
||
continue
|
||
|
||
# 等待服务器响应。这里只要数据全部 SEND OK，基本可认为上传成功；
|
||
# 响应读取主要用于调试和及时发现服务端拒绝。
|
||
print("[ESP] Waiting for response...")
|
||
resp = ""
|
||
start = utime.ticks_ms()
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < 3000:
|
||
if uart2.any():
|
||
data = uart2.read()
|
||
if data:
|
||
resp += data.decode('utf-8', 'ignore')
|
||
if "200" in resp or "success" in resp or "HTTP/1.1" in resp:
|
||
print("[ESP] Got response!")
|
||
break
|
||
utime.sleep_ms(50)
|
||
|
||
esp_send_cmd("AT+CIPCLOSE", 300)
|
||
|
||
# 数据发送成功就算成功
|
||
if send_ok:
|
||
print("[ESP] Photo uploaded successfully!")
|
||
return True
|
||
else:
|
||
print("[ESP] Response:", resp[:100] if resp else "(empty)")
|
||
print("[ESP] Retry {}/{}".format(attempt+1, max_retries))
|
||
utime.sleep_ms(1000)
|
||
|
||
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()
|
||
|
||
# 将当前画面压缩为 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:
|
||
print("[PHOTO] Compress method 2 failed:", e2)
|
||
return False
|
||
|
||
print("[PHOTO] JPEG size:", len(jpeg_data), "bytes")
|
||
|
||
if len(jpeg_data) < 100:
|
||
print("[PHOTO] JPEG too small, skip upload")
|
||
return False
|
||
|
||
# 先保存本地，再加入上传队列（非阻塞）。
|
||
# 注意：这里不直接调用 HTTP 上传，避免新增人脸时 UI 长时间卡住。
|
||
if not save_local_face_photo(face_id, jpeg_data):
|
||
return False
|
||
|
||
queue_photo_upload(face_id)
|
||
print("[PHOTO] Face {} saved locally and queued for upload".format(face_id))
|
||
return True
|
||
|
||
except Exception as e:
|
||
print("[PHOTO] Error:", e)
|
||
return False
|
||
|
||
# 初始化 WiFi 连接（启动时尝试连接）。
|
||
# 若启动时失败，识别功能仍然可用；照片会保存在 SD 卡，等后续重启或网络恢复后补传。
|
||
wifi_connected = False
|
||
print("[BOOT] Starting ESP8266 WiFi init...")
|
||
try:
|
||
wifi_connected = esp_init_wifi()
|
||
print("[BOOT] WiFi status:", "connected" if wifi_connected else "not connected")
|
||
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 卡上的映射文件，简单使用 key=value 文本格式
|
||
SYNC_VERSION_FILE = "/sd/sync_version.txt" # 本地已应用的服务端同步版本号
|
||
RECORD_PATH = "/sd/recordftr.txt" # 人脸特征记录文件路径
|
||
sync_last_version = 0 # K210 已处理到的服务端统一同步版本
|
||
|
||
# 函数作用：从本地文件加载人脸编号到学号的映射关系。
|
||
def load_local_mapping():
|
||
"""从SD卡加载本地映射配置"""
|
||
global face_id_mapping
|
||
face_id_mapping.clear()
|
||
try:
|
||
with open(MAPPING_FILE, "r") as f:
|
||
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]
|
||
print("[LOCAL] Loaded {} mappings from SD".format(len(face_id_mapping)))
|
||
except:
|
||
print("[LOCAL] No local mapping file")
|
||
|
||
# 函数作用：把当前人脸编号到学号映射保存到本地文件。
|
||
def save_local_mapping():
|
||
"""保存映射配置到SD卡"""
|
||
try:
|
||
with open(MAPPING_FILE, "w") as f:
|
||
for k, v in face_id_mapping.items():
|
||
f.write("{}={}\n".format(k, v))
|
||
print("[LOCAL] Saved {} mappings to SD".format(len(face_id_mapping)))
|
||
return True
|
||
except Exception as e:
|
||
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:
|
||
value = f.read().strip()
|
||
sync_last_version = int(value) if value else 0
|
||
except:
|
||
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:
|
||
with open(SYNC_VERSION_FILE, "w") as f:
|
||
f.write(str(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
|
||
|
||
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:
|
||
found = True
|
||
break
|
||
|
||
if found:
|
||
print("[UPDATE] Mapping saved: face_id {} -> student_id {}".format(face_id, new_student_id))
|
||
else:
|
||
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:
|
||
print("[HTTP_GET] WiFi未连接")
|
||
return None
|
||
|
||
print("[HTTP_GET] 关闭旧连接...")
|
||
esp_send_cmd("AT+CIPCLOSE", 500)
|
||
utime.sleep_ms(300)
|
||
esp_send_cmd("AT+CIPMUX=0", 500)
|
||
utime.sleep_ms(100)
|
||
|
||
# 建立TCP连接
|
||
print("[HTTP_GET] 建立TCP连接: {}:{}".format(SERVER_IP, SERVER_PORT))
|
||
cmd = 'AT+CIPSTART="TCP","{}",{}'.format(SERVER_IP, SERVER_PORT)
|
||
resp = esp_send_cmd(cmd, 5000)
|
||
print("[HTTP_GET] TCP响应: {}".format(resp.replace('\r\n', ' ')[:80] if resp else "(空)"))
|
||
|
||
if "OK" not in resp and "CONNECT" not in resp:
|
||
print("[HTTP_GET] TCP连接失败")
|
||
return None
|
||
|
||
print("[HTTP_GET] TCP连接成功")
|
||
utime.sleep_ms(200)
|
||
|
||
# 构建 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 发送请求头。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:
|
||
if uart2.any():
|
||
data = uart2.read()
|
||
if data:
|
||
prompt_resp += data.decode('utf-8', 'ignore')
|
||
if ">" in prompt_resp:
|
||
break
|
||
utime.sleep_ms(10)
|
||
|
||
if ">" not in prompt_resp:
|
||
print("[HTTP_GET] 未收到>提示符: {}".format(prompt_resp[:50]))
|
||
esp_send_cmd("AT+CIPCLOSE", 300)
|
||
return None
|
||
|
||
print("[HTTP_GET] 收到>提示符，发送请求...")
|
||
|
||
# 发送 HTTP 请求文本。
|
||
uart2.write(http_request)
|
||
|
||
# 等待 SEND OK，确认 ESP8266 已把请求写入 TCP 连接。
|
||
start = utime.ticks_ms()
|
||
send_resp = ""
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < 3000:
|
||
if uart2.any():
|
||
data = uart2.read()
|
||
if data:
|
||
send_resp += data.decode('utf-8', 'ignore')
|
||
if "SEND OK" in send_resp:
|
||
print("[HTTP_GET] SEND OK")
|
||
break
|
||
utime.sleep_ms(10)
|
||
|
||
if "SEND OK" not in send_resp:
|
||
print("[HTTP_GET] 发送失败: {}".format(send_resp[:50]))
|
||
esp_send_cmd("AT+CIPCLOSE", 300)
|
||
return None
|
||
|
||
# 等待服务器响应。
|
||
# 同步接口返回内容可能较长，所以不能只等第一次串口可读就结束。
|
||
print("[HTTP_GET] 等待服务器响应...")
|
||
resp = send_resp # 继续累积数据
|
||
start = utime.ticks_ms()
|
||
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < timeout:
|
||
if uart2.any():
|
||
data = uart2.read()
|
||
if data:
|
||
try:
|
||
resp += data.decode('utf-8', 'ignore')
|
||
print("[HTTP_GET] 累积数据: {} bytes".format(len(resp)))
|
||
except:
|
||
pass
|
||
# 加密响应是Base64文本，按Content-Length或连接关闭判断完整性。
|
||
if http_body_complete(resp):
|
||
if "CLOSED" in resp:
|
||
print("[HTTP_GET] 连接已关闭，数据完整")
|
||
break
|
||
# 继续等一下看有没有更多数据
|
||
utime.sleep_ms(500)
|
||
if uart2.any():
|
||
extra = uart2.read()
|
||
if extra:
|
||
resp += extra.decode('utf-8', 'ignore')
|
||
if "CLOSED" in resp or utime.ticks_diff(utime.ticks_ms(), start) > 3000:
|
||
break
|
||
utime.sleep_ms(30)
|
||
|
||
esp_send_cmd("AT+CIPCLOSE", 300)
|
||
|
||
print("[HTTP_GET] 原始响应长度: {}".format(len(resp)))
|
||
if len(resp) > 0:
|
||
print("[HTTP_GET] 响应片段: {}".format(resp[:300].replace('\r\n', '\\n')))
|
||
|
||
result = decode_http_json_body(resp)
|
||
if result:
|
||
print("[HTTP_GET] 解密JSON成功: {}".format(result[:100]))
|
||
return result
|
||
|
||
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()
|
||
|
||
# 启动时同步将在所有函数定义后执行（见文件末尾）
|
||
|
||
# --- 定时同步配置 ---
|
||
last_sync_time = utime.ticks_ms()
|
||
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)
|
||
|
||
deleted_index = -1
|
||
for i, p in enumerate(peoples):
|
||
for pattern in target_patterns:
|
||
if p.name == pattern:
|
||
deleted_index = i
|
||
break
|
||
if deleted_index >= 0:
|
||
break
|
||
|
||
if deleted_index >= 0:
|
||
deleted_name = peoples[deleted_index].name
|
||
peoples.pop(deleted_index)
|
||
try:
|
||
# 删除内存中的特征后立刻重写 recordftr.txt，保证下次启动不会恢复已删除人脸。
|
||
write_face_record()
|
||
print("[DELETE] 删除人脸成功: {}".format(deleted_name))
|
||
|
||
# 同时删除本地照片，避免下次启动补传已删除的人脸照片
|
||
delete_local_face_photo(face_id)
|
||
if deleted_name.startswith("ID:"):
|
||
delete_local_face_photo(deleted_name.split(":", 1)[1])
|
||
else:
|
||
delete_local_face_photo(deleted_name)
|
||
for pattern in target_patterns:
|
||
if pattern.startswith("ID:"):
|
||
delete_local_face_photo(pattern.split(":", 1)[1])
|
||
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:
|
||
print("[DELETE] 重写文件失败:", e)
|
||
load_faces()
|
||
return False
|
||
else:
|
||
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
|
||
|
||
esp_send_cmd("AT+CIPCLOSE", 300)
|
||
utime.sleep_ms(50)
|
||
esp_send_cmd("AT+CIPMUX=0", 300)
|
||
|
||
cmd = 'AT+CIPSTART="TCP","{}",{}'.format(SERVER_IP, SERVER_PORT)
|
||
resp = esp_send_cmd(cmd, 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)
|
||
|
||
if not esp_send_data(http_request, 1500):
|
||
esp_send_cmd("AT+CIPCLOSE", 300)
|
||
return None
|
||
|
||
resp = ""
|
||
start = utime.ticks_ms()
|
||
while utime.ticks_diff(utime.ticks_ms(), start) < timeout:
|
||
if uart2.any():
|
||
data = uart2.read()
|
||
if data:
|
||
resp += data.decode('utf-8', 'ignore')
|
||
if http_body_complete(resp):
|
||
break
|
||
utime.sleep_ms(10)
|
||
|
||
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:
|
||
return default
|
||
start += len(marker)
|
||
while start < len(text) and text[start] == ' ':
|
||
start += 1
|
||
end = start
|
||
while end < len(text) and text[end].isdigit():
|
||
end += 1
|
||
try:
|
||
return int(text[start:end])
|
||
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:
|
||
return ""
|
||
start = text.find(open_ch, pos + len(marker))
|
||
if start == -1:
|
||
return ""
|
||
depth = 0
|
||
for i in range(start, len(text)):
|
||
ch = text[i]
|
||
if ch == open_ch:
|
||
depth += 1
|
||
elif ch == close_ch:
|
||
depth -= 1
|
||
if depth == 0:
|
||
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):
|
||
start = arr_str.find('"', pos)
|
||
if start == -1:
|
||
break
|
||
end = arr_str.find('"', start + 1)
|
||
if end == -1:
|
||
break
|
||
values.append(arr_str[start + 1:end])
|
||
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):
|
||
obj_start = tasks_str.find('{', pos)
|
||
if obj_start == -1:
|
||
break
|
||
obj_end = tasks_str.find('}', obj_start)
|
||
if obj_end == -1:
|
||
break
|
||
obj = tasks_str[obj_start:obj_end + 1]
|
||
task_id = json_get_string(obj, "task_id", "")
|
||
face_id = json_get_string(obj, "face_id", "")
|
||
if face_id:
|
||
tasks.append({"task_id": task_id, "face_id": face_id})
|
||
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:
|
||
return default
|
||
start += len(marker)
|
||
end = text.find('"', start)
|
||
if end == -1:
|
||
return default
|
||
return text[start:end]
|
||
|
||
# 函数作用：从服务端照片列表响应中解析人脸照片编号。
|
||
def parse_photo_ids(response):
|
||
"""解析服务端 /api/v1/faces/photos 返回的照片ID列表"""
|
||
ids = {}
|
||
pos = 0
|
||
marker = '"id":"'
|
||
while pos < len(response):
|
||
start = response.find(marker, pos)
|
||
if start == -1:
|
||
break
|
||
start += len(marker)
|
||
end = response.find('"', start)
|
||
if end == -1:
|
||
break
|
||
photo_id = response[start:end]
|
||
if photo_id:
|
||
ids[photo_id] = 1
|
||
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
|
||
inner = obj_str[1:-1].strip()
|
||
if not inner:
|
||
return mappings
|
||
parts = inner.split(',')
|
||
for part in parts:
|
||
if ':' not in part:
|
||
continue
|
||
kv = part.split(':', 1)
|
||
key = kv[0].strip().strip('"')
|
||
value = kv[1].strip().strip('"')
|
||
if key and value:
|
||
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
|
||
print("[SYNC_STATE] 解除本地face_id绑定: ID:{}".format(face_id))
|
||
continue
|
||
print("[SYNC_STATE] 映射: ID:{} -> {}".format(face_id, student_id))
|
||
if update_face_record(face_id, student_id):
|
||
updated_count += 1
|
||
valid_mappings[face_id] = student_id
|
||
face_id_mapping = valid_mappings
|
||
save_local_mapping()
|
||
print("[SYNC_STATE] 映射应用完成，更新 {} 条".format(updated_count))
|
||
|
||
# 函数作用：从服务端同步统一状态，包括映射、照片和删除任务。
|
||
def sync_unified_state():
|
||
"""按服务端统一版本号同步K210所需状态"""
|
||
global sync_last_version
|
||
if not wifi_connected:
|
||
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)
|
||
if response is None:
|
||
print("[SYNC_STATE] 服务器无响应")
|
||
return False
|
||
|
||
server_version = json_get_int(response, "version", sync_last_version)
|
||
changed = ('"changed":true' in response)
|
||
print("[SYNC_STATE] local={}, server={}, changed={}".format(sync_last_version, server_version, changed))
|
||
|
||
if not changed:
|
||
# 服务端没有变化也保存版本号，防止本地版本落后但无需实际处理。
|
||
save_sync_version(server_version)
|
||
return True
|
||
|
||
profiles_arr = extract_json_block(response, "face_profiles", '[', ']')
|
||
server_mappings = {}
|
||
server_face_ids = {}
|
||
pos = 0
|
||
while profiles_arr and pos < len(profiles_arr):
|
||
obj_start = profiles_arr.find('{', pos)
|
||
if obj_start == -1:
|
||
break
|
||
obj_end = profiles_arr.find('}', obj_start)
|
||
if obj_end == -1:
|
||
break
|
||
obj = profiles_arr[obj_start:obj_end + 1]
|
||
face_id = json_get_string(obj, "face_id", "")
|
||
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__"
|
||
server_mappings[face_id] = student_id
|
||
pos = obj_end + 1
|
||
|
||
tasks_arr = extract_json_block(response, "delete_tasks", '[', ']')
|
||
delete_tasks = parse_delete_tasks(tasks_arr)
|
||
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 = []
|
||
delete_failed = False
|
||
print("[SYNC_STATE] 待删除任务: {}".format(delete_tasks))
|
||
|
||
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"]:
|
||
deleted_task_ids.append(task["task_id"])
|
||
continue
|
||
if delete_face_by_id(face_id):
|
||
deleted_ids.append(face_id)
|
||
if task["task_id"]:
|
||
deleted_task_ids.append(task["task_id"])
|
||
else:
|
||
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)
|
||
ack_resp = esp_http_post_json("/api/v1/sync/ack", ack_body, 3000)
|
||
print("[SYNC_STATE] ACK响应:", ack_resp[:120] if ack_resp else "(空)")
|
||
ack_compact = ack_resp.replace(" ", "").replace("\r", "").replace("\n", "") if ack_resp else ""
|
||
ack_ok = '"success":true' in ack_compact
|
||
if not ack_ok:
|
||
delete_failed = True
|
||
print("[SYNC_STATE] 删除ACK失败，暂不推进本地版本，等待下次重试")
|
||
|
||
# 删除任务处理完成后再应用映射，避免先写入后又删除造成短暂脏状态。
|
||
apply_face_profiles(server_mappings)
|
||
|
||
if not delete_failed:
|
||
save_sync_version(server_version)
|
||
else:
|
||
print("[SYNC_STATE] 有删除失败项，暂不推进本地版本，等待下次重试")
|
||
|
||
return not delete_failed
|
||
|
||
# --- K210时间管理 ---
|
||
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:
|
||
# 如果还没设置基准时间，使用运行时间
|
||
return utime.ticks_ms() // 1000
|
||
else:
|
||
# 计算当前时间戳
|
||
elapsed_seconds = (utime.ticks_ms() - k210_base_ticks) // 1000
|
||
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")
|
||
task_fe = kpu.load("/sd/FeatureExtraction.smodel")
|
||
except Exception as e:
|
||
print("Error loading KPU models:", e)
|
||
while True:
|
||
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)
|
||
LED_B = GPIO(GPIO.GPIO0, GPIO.OUT, value=1)
|
||
LED_G = GPIO(GPIO.GPIO1, GPIO.OUT, value=1)
|
||
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)
|
||
sensor.set_framesize(sensor.QVGA)
|
||
sensor.set_hmirror(CAMERA_HMIRROR)
|
||
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()
|
||
|
||
# --- 3. 按键中断处理 ---
|
||
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:
|
||
if key_Flag == 0:
|
||
key_Flag = 1
|
||
t_key = utime.ticks_ms()
|
||
else:
|
||
if key_Flag == 1:
|
||
press_duration = utime.ticks_diff(utime.ticks_ms(), t_key)
|
||
if press_duration > 5000:
|
||
key_pressed = 3 # 超长按5秒：删除全部人脸
|
||
elif press_duration > 2000:
|
||
key_pressed = 2 # 长按2秒：删除模式
|
||
else:
|
||
key_pressed = 1 # 短按：添加人脸
|
||
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()
|
||
print("[DELETE_ALL] 已删除全部人脸数据")
|
||
return True
|
||
except Exception as e:
|
||
print("[DELETE_ALL] 删除失败:", e)
|
||
load_faces()
|
||
return False
|
||
|
||
# --- 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':
|
||
return ord(ch) - ord('a') + 10
|
||
if 'A' <= ch <= 'F':
|
||
return ord(ch) - ord('A') + 10
|
||
return -1
|
||
|
||
# 函数作用：解析 Python bytes 字面量文本为字节值列表。
|
||
def parse_bytes_literal(text):
|
||
"""安全解析 bytearray(b"...") / b"..." 文本，不使用 eval。"""
|
||
pos1 = text.find("b'")
|
||
pos2 = text.find('b"')
|
||
if pos1 < 0 and pos2 < 0:
|
||
raise ValueError("bytes literal not found")
|
||
if pos1 >= 0 and (pos2 < 0 or pos1 < pos2):
|
||
pos = pos1
|
||
quote = "'"
|
||
else:
|
||
pos = pos2
|
||
quote = '"'
|
||
|
||
i = pos + 2
|
||
values = []
|
||
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")
|
||
esc = text[i]
|
||
if esc == "x":
|
||
if i + 2 >= len(text):
|
||
raise ValueError("bad hex escape")
|
||
hi = hex_value(text[i + 1])
|
||
lo = hex_value(text[i + 2])
|
||
if hi < 0 or lo < 0:
|
||
raise ValueError("bad hex escape")
|
||
values.append(hi * 16 + lo)
|
||
i += 3
|
||
elif esc == "n":
|
||
values.append(10)
|
||
i += 1
|
||
elif esc == "r":
|
||
values.append(13)
|
||
i += 1
|
||
elif esc == "t":
|
||
values.append(9)
|
||
i += 1
|
||
elif esc == "b":
|
||
values.append(8)
|
||
i += 1
|
||
elif esc == "f":
|
||
values.append(12)
|
||
i += 1
|
||
elif esc == "v":
|
||
values.append(11)
|
||
i += 1
|
||
elif esc == "a":
|
||
values.append(7)
|
||
i += 1
|
||
elif esc == "\\":
|
||
values.append(92)
|
||
i += 1
|
||
elif esc == quote:
|
||
values.append(ord(quote))
|
||
i += 1
|
||
elif '0' <= esc <= '7':
|
||
val = ord(esc) - ord('0')
|
||
i += 1
|
||
count = 1
|
||
while count < 3 and i < len(text) and '0' <= text[i] <= '7':
|
||
val = val * 8 + ord(text[i]) - ord('0')
|
||
i += 1
|
||
count += 1
|
||
values.append(val & 0xff)
|
||
else:
|
||
values.append(ord(esc))
|
||
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:
|
||
raise ValueError("feature must be bytearray/list")
|
||
|
||
inner = text[start + 1:end].strip()
|
||
if not inner:
|
||
raise ValueError("empty feature")
|
||
|
||
values = []
|
||
has_float = False
|
||
all_byte = True
|
||
for token in inner.split(","):
|
||
token = token.strip()
|
||
if not token:
|
||
raise ValueError("empty number")
|
||
if "." in token or "e" in token or "E" in token:
|
||
value = float(token)
|
||
has_float = True
|
||
all_byte = False
|
||
else:
|
||
value = int(token)
|
||
if value < 0 or value > 255:
|
||
all_byte = False
|
||
values.append(value)
|
||
if len(values) > 4096:
|
||
raise ValueError("feature too long")
|
||
|
||
if len(values) < 64:
|
||
raise ValueError("feature too short")
|
||
if all_byte and not has_float:
|
||
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)
|
||
return parse_number_list(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
|
||
if v == 92:
|
||
out += "\\\\"
|
||
elif v == 34:
|
||
out += '\\"'
|
||
elif v == 10:
|
||
out += "\\n"
|
||
elif v == 13:
|
||
out += "\\r"
|
||
elif v == 9:
|
||
out += "\\t"
|
||
elif 32 <= v <= 126:
|
||
out += chr(v)
|
||
else:
|
||
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 = []
|
||
for v in feature:
|
||
values.append(int(v) & 0xff)
|
||
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
|
||
print("[FACE_DB] Loading {}".format(record_path))
|
||
try:
|
||
with open(record_path, "r") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
total_count += 1
|
||
# 使用 maxsplit=1 参数，只按第一个 '#' 分割，防止特征数据中的'#'导致错误
|
||
parts = line.split('#', 1)
|
||
if len(parts) == 2 and parts[1]:
|
||
try:
|
||
peoples.append(People(parts[0], parse_feature_vector(parts[1])))
|
||
except Exception as e_eval:
|
||
bad_count += 1
|
||
print("Feature parse error on line {} for {}: {}".format(total_count, parts[0], e_eval))
|
||
else:
|
||
bad_count += 1
|
||
print("[FACE_DB] Bad line {}: missing feature".format(total_count))
|
||
if total_count == 0:
|
||
print("[FACE_DB] recordftr.txt is empty")
|
||
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):
|
||
if filename.endswith(".jpg"):
|
||
face_id = filename[:-4]
|
||
if face_id.isdigit():
|
||
used[face_id] = 1
|
||
except Exception as e:
|
||
print("[ADD] 扫描本地照片ID失败:", e)
|
||
|
||
return used
|
||
|
||
# 函数作用：分配下一个可用的人脸编号。
|
||
def allocate_next_face_id():
|
||
# 从 1 开始寻找最小可用 face_id，便于现场查看和管理端展示。
|
||
used = collect_used_face_ids()
|
||
next_id = 1
|
||
while str(next_id) in used:
|
||
next_id += 1
|
||
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))
|
||
try:
|
||
if face_img is not None:
|
||
print("[PHOTO] 正在保存人脸照片并加入上传队列...")
|
||
if not capture_and_upload_face(new_id_num, face_img):
|
||
print("[PHOTO] 本地照片保存失败，取消添加人脸")
|
||
peoples.pop()
|
||
return False
|
||
|
||
with open(record_path, "a") as f:
|
||
f.write(new_name + '#' + feature_to_record_text(feature) + "\n")
|
||
print("添加成功:", new_name)
|
||
|
||
return True
|
||
except Exception as e:
|
||
print("写入文件失败:", e)
|
||
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)
|
||
try:
|
||
write_face_record()
|
||
if deleted_name.startswith("ID:"):
|
||
delete_local_face_photo(deleted_name.split(":", 1)[1])
|
||
else:
|
||
delete_local_face_photo(deleted_name)
|
||
print("删除 %s 成功" % deleted_name); return True
|
||
except Exception as e: print("重写文件失败:", e); load_faces(); return False
|
||
|
||
# --- 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 = []
|
||
for item in codes:
|
||
face_cut = img.cut(item.x(), item.y(), item.w(), item.h())
|
||
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)
|
||
return results
|
||
|
||
# --- 6. 主程序启动前准备 - 添加发送逻辑所需变量 ---
|
||
load_faces()
|
||
last_send_time = 0
|
||
SEND_INTERVAL_MS = 1000 # 串口上报最小间隔，避免同一张脸每帧都发送给 STM32
|
||
last_sent_data = "" # 记录上一次完整串口消息，内容不变时不重复发送
|
||
last_saved_id = "" # 预留字段：记录最近一次保存/处理的学号，切换状态时重置
|
||
|
||
# --- 分时网络：照片上传队列操作 ---
|
||
# 函数作用：把指定人脸编号加入照片待上传队列。
|
||
def queue_photo_upload(face_id):
|
||
"""将本地照片ID添加到上传队列（非阻塞）"""
|
||
global photo_upload_queue
|
||
face_id = str(face_id)
|
||
if face_id in photo_upload_queue:
|
||
print("[QUEUE] 照片已在队列中: ID={}".format(face_id))
|
||
return
|
||
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
|
||
|
||
server_photo_ids = None
|
||
if wifi_connected:
|
||
print("[QUEUE] 正在获取服务端照片列表，用于判断缺失照片...")
|
||
response = esp_http_get("/api/v1/faces/photos", 8000)
|
||
if response:
|
||
server_photo_ids = parse_photo_ids(response)
|
||
print("[QUEUE] 服务端已有照片: {} 张".format(len(server_photo_ids)))
|
||
else:
|
||
print("[QUEUE] 服务端照片列表获取失败，将本地照片全部加入补传队列")
|
||
|
||
queued = 0
|
||
skipped = 0
|
||
try:
|
||
for name in os.listdir(LOCAL_FACE_PHOTO_DIR):
|
||
if name.endswith(".jpg"):
|
||
face_id = name[:-4]
|
||
if face_id:
|
||
if server_photo_ids is not None:
|
||
server_has_photo = face_id in server_photo_ids
|
||
if server_has_photo:
|
||
skipped += 1
|
||
continue
|
||
queue_photo_upload(face_id)
|
||
queued += 1
|
||
print("[QUEUE] 本地照片补传扫描完成: queued={}, skipped={}".format(queued, skipped))
|
||
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)
|
||
if jpeg_data is None or len(jpeg_data) < 100:
|
||
print("[NET] 本地照片无效，跳过上传: ID={}".format(face_id))
|
||
return True
|
||
|
||
print("[NET] 上传照片: ID={}".format(face_id))
|
||
try:
|
||
if esp_http_post_photo(face_id, jpeg_data):
|
||
print("[NET] 照片上传完成")
|
||
else:
|
||
print("[NET] 照片上传失败，保留队列等待重试")
|
||
queue_photo_upload(face_id)
|
||
except Exception as e:
|
||
print("[NET] 上传错误:", e)
|
||
queue_photo_upload(face_id)
|
||
return True
|
||
|
||
return False
|
||
|
||
# 启动时扫描本地照片，补传可能遗漏到服务器的照片
|
||
queue_local_photos_for_upload()
|
||
|
||
# 启动时执行一次同步
|
||
print("[BOOT] 启动同步...")
|
||
if wifi_connected:
|
||
try:
|
||
sync_unified_state()
|
||
except Exception as e:
|
||
print("[BOOT] 统一同步错误:", e)
|
||
print("[BOOT] 同步完成")
|
||
else:
|
||
print("[BOOT] WiFi未连接，跳过同步")
|
||
|
||
# QT触发同步标志
|
||
sync_requested = False # 服务端提示需要同步后置为 True，主循环空闲时执行 trigger_sync
|
||
last_sync_check_time = 0 # 上一次检查服务端同步请求的时间
|
||
|
||
# 函数作用：检查服务端是否要求 K210 立即同步。
|
||
def check_sync_request():
|
||
"""检查服务器是否有同步请求"""
|
||
global sync_requested
|
||
if not wifi_connected:
|
||
return False
|
||
|
||
try:
|
||
response = esp_http_get("/api/v1/sync/request", 2000)
|
||
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
|
||
if not wifi_connected:
|
||
print("[SYNC] WiFi未连接，无法同步")
|
||
return False
|
||
|
||
print("[SYNC] ========== QT触发同步开始 ==========")
|
||
try:
|
||
sync_unified_state()
|
||
except Exception as e:
|
||
print("[SYNC] 统一同步错误:", e)
|
||
|
||
print("[SYNC] ========== QT触发同步完成 ==========")
|
||
sync_requested = False
|
||
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
|
||
for face in detected_faces:
|
||
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，STM32 可据此提示未注册或拒绝签到。
|
||
msg_to_send = "UNKNOWN\r\n"
|
||
|
||
# 默认识别模式下陌生人不自动入库，只有进入添加模式后才保存。
|
||
last_saved_id = "" # 重置保存记录
|
||
else:
|
||
# 已识别用户发送 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:
|
||
print("[UART SEND] Face {} has no student mapping, fallback to face_id".format(face_id))
|
||
student_id = mapped_student_id if mapped_student_id else face_id
|
||
student_name = mapped_student_id if mapped_student_id else "FACE_{}".format(face_id)
|
||
confidence = best_face['score'] / 100.0
|
||
|
||
# 构造 STM32 期望的格式；未绑定学号时默认使用 K210 原始 face_id。
|
||
msg_to_send = "ID:{},NAME:{},CONF:{:.2f}\r\n".format(
|
||
student_id,
|
||
student_name,
|
||
confidence
|
||
)
|
||
|
||
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 = ""
|
||
last_saved_id = "" # 重置保存记录
|
||
last_send_time = time_now
|
||
lcd.display(img)
|
||
|
||
# ------------------ 模式1：添加人脸 ------------------
|
||
elif key_pressed == 1:
|
||
key_pressed = 0
|
||
# 进入管理模式前清空上报状态，退出后下一次识别会重新发送。
|
||
last_sent_data = "" # 修改变量名
|
||
last_saved_id = "" # 重置保存记录
|
||
last_send_time = 0
|
||
print("进入添加人脸模式...")
|
||
while True:
|
||
img = sensor.snapshot()
|
||
img.draw_string(0, 0, "Add Face Mode", color=(0, 0, 255), scale=2)
|
||
img.draw_string(0, 24, "Click to Confirm", color=(0, 0, 255), scale=2)
|
||
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:
|
||
img.draw_string(best_face['rect'][0], best_face['rect'][1] + 24, "Add Failed!", color=(255, 0, 0), scale=2); LED_R.value(0)
|
||
else:
|
||
img.draw_string(10, 100, "No Face!", color=(255, 0, 0), scale=2); LED_R.value(0)
|
||
lcd.display(img); utime.sleep_ms(2000); LED_R.value(1); LED_G.value(1); break
|
||
if key_pressed == 2:
|
||
key_pressed = 0; break # 长按退出
|
||
lcd.display(img)
|
||
print("退出添加人脸模式")
|
||
|
||
# ------------------ 模式2：删除人脸 ------------------
|
||
elif key_pressed == 2:
|
||
key_pressed = 0
|
||
last_sent_data = "" # 修改变量名
|
||
last_saved_id = "" # 重置保存记录
|
||
last_send_time = 0
|
||
print("进入删除人脸模式...")
|
||
select_index = -1
|
||
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):
|
||
img.clear(); img.draw_string(50, 100, "Delete Success!", color=(0, 255, 0), scale=2); LED_G.value(0)
|
||
else:
|
||
img.clear(); img.draw_string(50, 100, "Delete Failed!", color=(255, 0, 0), scale=2); LED_R.value(0)
|
||
else:
|
||
img.clear(); img.draw_string(50, 100, "Canceled", color=(255, 255, 0), scale=2)
|
||
lcd.display(img); utime.sleep_ms(2000); LED_G.value(1); LED_R.value(1); load_faces(); break
|
||
lcd.display(img)
|
||
print("退出删除人脸模式")
|
||
|
||
# ------------------ 模式3：删除全部人脸 (长按5秒) ------------------
|
||
elif key_pressed == 3:
|
||
key_pressed = 0
|
||
print("进入删除全部人脸模式...")
|
||
img = image.Image(); img.clear()
|
||
img.draw_string(10, 10, "DELETE ALL FACES?", color=(255, 0, 0), scale=2)
|
||
img.draw_string(10, 50, "Count: {}".format(len(peoples)), color=(255, 255, 0), scale=2)
|
||
img.draw_string(10, 90, "Click=Confirm", color=(255, 128, 0), scale=1.8)
|
||
img.draw_string(10, 120, "Long Press=Cancel", color=(0, 255, 0), scale=1.8)
|
||
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: # 短按确认删除
|
||
key_pressed = 0
|
||
img.clear()
|
||
img.draw_string(30, 80, "Deleting...", color=(255, 0, 0), scale=2)
|
||
lcd.display(img)
|
||
if delete_all_faces():
|
||
img.clear()
|
||
img.draw_string(20, 80, "All Deleted!", color=(0, 255, 0), scale=2)
|
||
LED_R.value(1); LED_G.value(0)
|
||
else:
|
||
img.clear()
|
||
img.draw_string(20, 80, "Delete Failed!", color=(255, 0, 0), scale=2)
|
||
lcd.display(img)
|
||
utime.sleep_ms(2000)
|
||
break
|
||
elif key_pressed == 2 or key_pressed == 3: # 长按取消
|
||
key_pressed = 0
|
||
img.clear()
|
||
img.draw_string(50, 80, "Canceled", color=(255, 255, 0), scale=2)
|
||
lcd.display(img)
|
||
utime.sleep_ms(1500)
|
||
break
|
||
utime.sleep_ms(50)
|
||
|
||
LED_R.value(1); LED_G.value(1)
|
||
print("退出删除全部人脸模式")
|
||
|
||
# 分时网络处理：每N帧检查一次网络任务（只处理照片上传和同步请求）
|
||
# 网络调度放在所有模式处理之后执行。
|
||
# 每次只做一类任务，优先级为：
|
||
# 1. 上传本地照片；
|
||
# 2. 执行已收到的同步请求；
|
||
# 3. 定时向服务端询问是否需要同步。
|
||
frames_since_network += 1
|
||
if frames_since_network >= NETWORK_FRAME_INTERVAL:
|
||
frames_since_network = 0
|
||
# 优先处理照片上传队列
|
||
if len(photo_upload_queue) > 0:
|
||
do_network_task()
|
||
# 检查是否有同步请求待执行
|
||
elif sync_requested:
|
||
trigger_sync()
|
||
# 定期检查服务器是否有新的同步请求
|
||
elif wifi_connected and utime.ticks_diff(utime.ticks_ms(), last_sync_check_time) > SYNC_CHECK_INTERVAL_MS:
|
||
last_sync_check_time = utime.ticks_ms()
|
||
check_sync_request()
|
||
|
||
gc.collect()
|
||
|