yzfh/bishe-K210
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加加密

Browse Source
This commit is contained in:
纪田野
2026-05-04 21:02:28 +08:00
parent 40def5153a
commit 85da30a6dd
1 changed files with 395 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+394 -286
View File
@@ -14,6 +14,17 @@ 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，默认堆较小，增加到512KB用于图像压缩
@@ -41,6 +52,125 @@ 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" # 本地人脸照片缓存目录
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"
def crypto_ready():
return crypto_lib is not None and ubinascii is not None
def pkcs7_pad(data):
pad_len = 16 - (len(data) % 16)
return data + bytes([pad_len]) * pad_len
def pkcs7_unpad(data):
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]
def aes_cbc_crypt(data, encrypt=True):
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
def encrypt_payload_base64(data):
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")
def decrypt_payload_base64(data):
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
def extract_http_body(resp):
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()
def http_body_complete(resp):
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
def decode_http_json_body(resp):
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
@@ -519,6 +649,10 @@ def update_face_record(face_id, new_student_id):
face_id = str(face_id)
new_student_id = str(new_student_id)
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()
@@ -560,7 +694,7 @@ def esp_http_get(url_path, timeout=8000):
utime.sleep_ms(200)
# 构建HTTP GET请求
http_request = "GET {} HTTP/1.1\r\nHost: {}:{}\r\nX-Api-Token: {}\r\nConnection: close\r\n\r\n".format(
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发送
@@ -623,8 +757,8 @@ def esp_http_get(url_path, timeout=8000):
print("[HTTP_GET] 累积数据: {} bytes".format(len(resp)))
except:
pass
# 检查是否收到完整响应（有JSON且连接关闭）
if "}" in resp and "{" in resp:
# 加密响应是Base64文本，按Content-Length或连接关闭判断完整性。
if http_body_complete(resp):
if "CLOSED" in resp:
print("[HTTP_GET] 连接已关闭，数据完整")
break
@@ -644,103 +778,27 @@ def esp_http_get(url_path, timeout=8000):
if len(resp) > 0:
print("[HTTP_GET] 响应片段: {}".format(resp[:300].replace('\r\n', '\\n')))
# 提取JSON
json_start = resp.find('{"')
if json_start != -1:
json_end = resp.rfind('}')
if json_end != -1 and json_end > json_start:
result = resp[json_start:json_end+1]
print("[HTTP_GET] 提取JSON成功: {}".format(result[:100]))
result = decode_http_json_body(resp)
if result:
print("[HTTP_GET] 解密JSON成功: {}".format(result[:100]))
return result
print("[HTTP_GET] 未找到JSON")
print("[HTTP_GET] 未找到可用JSON")
return None
def sync_face_id_mapping():
"""从服务器同步人脸ID映射配置并更新本地记录"""
global face_id_mapping
if not wifi_connected:
print("[SYNC] WiFi not connected, skip sync")
return False
try:
print("[SYNC] ========== 开始同步映射配置 ==========")
print("[SYNC] 请求: GET /api/v1/faces/mappings")
response = esp_http_get("/api/v1/faces/mappings", 2000)
if response is None:
print("[SYNC] 服务器无响应")
return False
print("[SYNC] 收到响应: {}".format(response[:150] if len(response) > 150 else response))
# 简单解析JSON
mappings_start = response.find('"mappings":{')
if mappings_start == -1:
print("[SYNC] 响应中无mappings字段")
return False
mappings_start += len('"mappings":{')
mappings_end = response.find('}', mappings_start)
if mappings_end == -1:
return False
mappings_str = response[mappings_start:mappings_end]
print("[SYNC] 解析映射字符串: {}".format(mappings_str))
# 解析键值对
server_mappings = {}
parts = mappings_str.split(',')
for part in parts:
part = part.strip()
if ':' in part:
kv = part.split(':')
if len(kv) == 2:
key = kv[0].strip().strip('"')
value = kv[1].strip().strip('"')
if key and value:
server_mappings[key] = value
print("[SYNC] 服务器映射数量: {}".format(len(server_mappings)))
for k, v in server_mappings.items():
print("[SYNC] 人脸ID: {} -> 学号: {}".format(k, v))
if len(server_mappings) == 0:
print("[SYNC] 服务器无映射配置")
face_id_mapping = {}
save_local_mapping()
return True
# 对比本地映射
print("[SYNC] 本地映射数量: {}".format(len(face_id_mapping)))
for k, v in face_id_mapping.items():
print("[SYNC] 本地: {} -> {}".format(k, v))
# 只更新 face_id -> 学号映射，保持 recordftr.txt 里的原始 face_id 稳定。
updated_count = 0
for face_id, student_id in server_mappings.items():
print("[SYNC] 检查是否需要更新: ID:{} -> {}".format(face_id, student_id))
if update_face_record(face_id, student_id):
updated_count += 1
# 更新内存中的映射表
face_id_mapping = server_mappings
save_local_mapping()
print("[SYNC] ========== 同步完成，更新了 {} 条记录 ==========".format(updated_count))
return True
except Exception as e:
print("[SYNC] 同步出错:", e)
return False
def extract_face_id(face_name):
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 = 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 = extract_face_id(face_name)
return face_id_mapping.get(face_id, "")
@@ -783,9 +841,7 @@ def delete_face_by_id(face_id):
deleted_name = peoples[deleted_index].name
peoples.pop(deleted_index)
try:
with open(record_path, "w") as f:
for p in peoples:
f.write(p.name + '#' + str(p.feature) + "\n")
write_face_record()
print("[DELETE] 删除人脸成功: {}".format(deleted_name))
# 同时删除本地照片，避免下次启动补传已删除的人脸照片
@@ -820,6 +876,9 @@ def esp_http_post_json(url_path, json_data, timeout=2000):
"""发送HTTP POST请求（优化版）"""
if not wifi_connected:
return None
encrypted_data = encrypt_payload_base64(json_data)
if encrypted_data is None:
return None
esp_send_cmd("AT+CIPCLOSE", 300)
utime.sleep_ms(50)
@@ -830,8 +889,8 @@ 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/json\r\nX-Api-Token: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}".format(
url_path, SERVER_IP, SERVER_PORT, API_TOKEN, len(json_data), json_data)
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)
@@ -844,164 +903,12 @@ def esp_http_post_json(url_path, json_data, timeout=2000):
data = uart2.read()
if data:
resp += data.decode('utf-8', 'ignore')
if "success" in resp or "}" in resp:
if http_body_complete(resp):
break
utime.sleep_ms(10)
esp_send_cmd("AT+CIPCLOSE", 300)
return resp
def sync_pending_delete():
"""从服务器同步待删除的人脸列表"""
if not wifi_connected:
print("[DELETE_SYNC] WiFi未连接，跳过")
return False
try:
print("[DELETE_SYNC] ========== 开始同步删除列表 ==========")
print("[DELETE_SYNC] 请求: GET /api/v1/faces/pending-delete")
response = esp_http_get("/api/v1/faces/pending-delete", 8000)
if response is None:
print("[DELETE_SYNC] 服务器无响应")
return False
print("[DELETE_SYNC] 收到响应: {}".format(response))
# 解析JSON
start = response.find('"pending_delete":[')
if start == -1:
print("[DELETE_SYNC] 响应中无pending_delete字段")
return True
start += len('"pending_delete":[')
end = response.find(']', start)
if end == -1:
return False
arr_str = response[start:end]
print("[DELETE_SYNC] 待删除列表字符串: '{}'".format(arr_str))
if not arr_str.strip():
print("[DELETE_SYNC] 无待删除的人脸")
return True
# 解析数组
ids_to_delete = []
parts = arr_str.split(',')
for part in parts:
part = part.strip().strip('"')
if part:
ids_to_delete.append(part)
print("[DELETE_SYNC] 待删除人脸ID列表: {}".format(ids_to_delete))
if not ids_to_delete:
print("[DELETE_SYNC] 解析后无待删除项")
return True
# 显示当前本地人脸列表
print("[DELETE_SYNC] 当前本地人脸数量: {}".format(len(peoples)))
for i, p in enumerate(peoples):
print("[DELETE_SYNC] [{}] {}".format(i, p.name))
# 执行删除（每次最多处理一个，避免长时间阻塞）
face_id = ids_to_delete[0]
print("[DELETE_SYNC] 正在删除: {}".format(face_id))
delete_result = delete_face_by_id(face_id)
print("[DELETE_SYNC] 删除结果: {}".format("成功" if delete_result else "失败"))
if delete_result:
# 只有本地特征真正删除成功后才确认，避免服务端过早清掉待删除任务
print("[DELETE_SYNC] 发送确认: POST /api/v1/faces/delete-ack")
confirm_result = esp_http_post_confirm(face_id)
print("[DELETE_SYNC] 确认结果: {}".format("成功" if confirm_result else "失败"))
else:
print("[DELETE_SYNC] 本地未删除成功，暂不确认，等待下次同步重试")
# 显示删除后的人脸列表
print("[DELETE_SYNC] 删除后本地人脸数量: {}".format(len(peoples)))
for i, p in enumerate(peoples):
print("[DELETE_SYNC] [{}] {}".format(i, p.name))
print("[DELETE_SYNC] ========== 删除同步完成 ==========")
return True
except Exception as e:
print("[DELETE_SYNC] 同步出错:", e)
return False
def esp_http_post_confirm(face_id):
"""发送删除确认请求"""
if not wifi_connected:
return False
try:
esp_send_cmd("AT+CIPCLOSE", 500)
utime.sleep_ms(200)
esp_send_cmd("AT+CIPMUX=0", 500)
# 建立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("[CONFIRM] TCP连接失败")
return False
# 构建POST请求
json_data = '{{"face_id":"{}"}}'.format(face_id)
http_request = "POST /api/v1/faces/delete-ack HTTP/1.1\r\n"
http_request += "Host: {}:{}\r\n".format(SERVER_IP, SERVER_PORT)
http_request += "Content-Type: application/json\r\n"
http_request += "X-Api-Token: {}\r\n".format(API_TOKEN)
http_request += "Content-Length: {}\r\n".format(len(json_data))
http_request += "Connection: close\r\n\r\n"
http_request += json_data
# 发送请求
cmd = "AT+CIPSEND={}".format(len(http_request))
uart2.write(cmd + "\r\n")
utime.sleep_ms(100)
# 等待>提示符
start = utime.ticks_ms()
prompt = ""
while utime.ticks_diff(utime.ticks_ms(), start) < 2000:
if uart2.any():
prompt += uart2.read().decode('utf-8', 'ignore')
if ">" in prompt:
break
utime.sleep_ms(10)
if ">" not in prompt:
esp_send_cmd("AT+CIPCLOSE", 300)
return False
# 发送数据
uart2.write(http_request)
# 等待响应
resp = ""
start = utime.ticks_ms()
while utime.ticks_diff(utime.ticks_ms(), start) < 5000:
if uart2.any():
data = uart2.read()
if data:
resp += data.decode('utf-8', 'ignore')
if "success" in resp:
print("[CONFIRM] 服务器确认成功")
esp_send_cmd("AT+CIPCLOSE", 300)
return True
utime.sleep_ms(20)
esp_send_cmd("AT+CIPCLOSE", 300)
print("[CONFIRM] 响应: {}".format(resp[:100] if resp else "(空)"))
return "success" in resp
except Exception as e:
print("[CONFIRM] 错误:", e)
return False
return decode_http_json_body(resp)
def json_get_int(text, key, default=0):
marker = '"{}":'.format(key)
@@ -1118,15 +1025,33 @@ def parse_mapping_object(obj_str):
mappings[key] = value
return mappings
def apply_face_mappings(server_mappings):
def apply_face_profiles(server_mappings):
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__":
print("[SYNC_STATE] 服务端仅有照片占位，保留本地映射: ID:{}".format(face_id))
continue
if not 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
face_id_mapping = server_mappings
valid_mappings[face_id] = student_id
face_id_mapping = valid_mappings
save_local_mapping()
print("[SYNC_STATE] 映射应用完成，更新 {} 条".format(updated_count))
@@ -1152,6 +1077,28 @@ def sync_unified_state():
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:
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", '[', ']')
@@ -1165,6 +1112,18 @@ def sync_unified_state():
for task in delete_tasks:
face_id = task["face_id"]
if face_id in server_face_ids:
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):
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"]:
@@ -1184,9 +1143,7 @@ def sync_unified_state():
delete_failed = True
print("[SYNC_STATE] 删除ACK失败，暂不推进本地版本，等待下次重试")
mappings_obj = extract_json_block(response, "face_mappings", '{', '}')
server_mappings = parse_mapping_object(mappings_obj)
apply_face_mappings(server_mappings)
apply_face_profiles(server_mappings)
if not delete_failed:
save_sync_version(server_version)
@@ -1301,45 +1258,205 @@ class People:
peoples = []
record_path = "/sd/recordftr.txt"
def parse_feature_vector(text):
"""只解析浮点数组文本，避免执行SD卡中的任意代码。"""
allowed = "0123456789+-eE.,[] \t\r\n"
for ch in text:
if ch not in allowed:
raise ValueError("invalid char in feature")
def hex_value(ch):
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
text = text.strip()
if not text.startswith("[") or not text.endswith("]"):
raise ValueError("feature must be list")
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 = '"'
inner = text[1:-1].strip()
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 != "\\":
values.append(ord(ch))
i += 1
else:
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:
raise ValueError("feature too long")
raise ValueError("unterminated bytes literal")
def parse_number_list(text):
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")
values.append(float(token))
if len(values) > 512:
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):
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):
return HEX_CHARS[(v >> 4) & 15] + HEX_CHARS[v & 15]
def escape_byte_values(values):
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):
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)
def write_face_record():
with open(record_path, "w") as f:
for p in peoples:
f.write(str(p.name) + '#' + feature_to_record_text(p.feature) + "\n")
def load_faces():
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.strip().split('#', 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: print("Feature parse error on line for {}: {}".format(parts[0], e_eval))
print("成功加载 %d 条人脸数据" % len(peoples))
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():
@@ -1388,7 +1505,8 @@ def add_face(feature, face_img=None):
peoples.pop()
return False
with open(record_path, "a") as f: f.write(new_name + '#' + str(feature) + "\n")
with open(record_path, "a") as f:
f.write(new_name + '#' + feature_to_record_text(feature) + "\n")
print("添加成功:", new_name)
return True
@@ -1402,8 +1520,7 @@ def delete_face_by_index(idx):
deleted_name = peoples[idx].name
peoples.pop(idx)
try:
with open(record_path, "w") as f:
for p in peoples: f.write(p.name + '#' + str(p.feature) + "\n")
write_face_record()
if deleted_name.startswith("ID:"):
delete_local_face_photo(deleted_name.split(":", 1)[1])
else:
@@ -1612,21 +1729,12 @@ while True:
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, send UNKNOWN".format(face_id))
msg_to_send = "UNKNOWN\r\n"
last_saved_id = ""
if msg_to_send != last_sent_data:
uart.write(msg_to_send)
print("[UART SEND] -> {}".format(msg_to_send.strip()))
last_sent_data = msg_to_send
last_send_time = time_now
lcd.display(img)
continue
student_id = mapped_student_id
student_name = 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期望的格式：ID:<学号>,NAME:<学号>,CONF:0.95
# 构造STM32期望的格式；未绑定学号时默认使用K210原始face_id。
msg_to_send = "ID:{},NAME:{},CONF:{:.2f}\r\n".format(
student_id,
student_name,