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

硬件基本完成

Browse Source
This commit is contained in:
纪田野
2026-05-01 21:34:10 +08:00
parent df5b05e936
commit 5a31ff2179
14 changed files with 1005 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files
k210/main.py
+116 -11
View File
@@ -35,7 +35,7 @@ WIFI_SSID = "yzfh" # 修改为你的WiFi名称
WIFI_PASSWORD = "yzfhroot" # 修改为你的WiFi密码
SERVER_IP = "27.106.113.205" # 修改为服务器IP地址
SERVER_PORT = 8082 # 服务器HTTP端口
API_TOKEN = "change-me-token" # 必须与服务端 ATTENDANCE_API_TOKEN 一致
API_TOKEN = "atk_nKfVWLhMjL0RGSfRll3R_sdJS6IrIjuV" # 必须与服务端 ATTENDANCE_API_TOKEN 一致
ESP_DEBUG = True # 打印ESP8266 AT响应，便于排查WiFi连接问题
ESP_BAUD_RATES = (115200, 9600, 74880, 57600, 38400)
ESP_UART_BAUD = 115200
@@ -1104,6 +1104,35 @@ def parse_string_array_block(arr_str):
pos = end + 1
return values
def parse_delete_tasks(tasks_str):
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
def json_get_string(text, key, default=""):
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 = {}
@@ -1175,23 +1204,37 @@ def sync_unified_state():
save_sync_version(server_version)
return True
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:
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(pending_delete))
print("[SYNC_STATE] 待删除任务: {}".format(delete_tasks))
for face_id in pending_delete:
for task in delete_tasks:
face_id = task["face_id"]
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:
deleted_json = ",".join(['"{}"'.format(x) for x in deleted_ids])
ack_body = '{{"client":"k210","version":{},"deleted":[{}]}}'.format(server_version, deleted_json)
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失败，暂不推进本地版本，等待下次重试")
mappings_obj = extract_json_block(response, "face_mappings", '{', '}')
server_mappings = parse_mapping_object(mappings_obj)
@@ -1309,6 +1352,35 @@ class People:
return kpu.face_compare(self.feature, feature)
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")
text = text.strip()
if not text.startswith("[") or not text.endswith("]"):
raise ValueError("feature must be list")
inner = text[1:-1].strip()
if not inner:
raise ValueError("empty feature")
values = []
for token in inner.split(","):
token = token.strip()
if not token:
raise ValueError("empty number")
values.append(float(token))
if len(values) > 512:
raise ValueError("feature too long")
if len(values) < 64:
raise ValueError("feature too short")
return values
def load_faces():
global peoples; peoples = []
try:
@@ -1317,16 +1389,49 @@ def load_faces():
# 使用 maxsplit=1 参数，只按第一个 '#' 分割，防止特征数据中的'#'导致错误
parts = line.strip().split('#', 1)
if len(parts) == 2 and parts[1]:
try: peoples.append(People(parts[0], eval(parts[1])))
except Exception as e_eval: print("Eval error on line for {}: {}".format(parts[0], e_eval))
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))
except Exception as e: print("读取人脸数据库失败:", e)
def collect_used_face_ids():
"""收集本地已经占用的K210原始face_id，避免新增时覆盖旧人脸。"""
used = {}
for p in peoples:
name = str(p.name)
if name.startswith("ID:") and name.split(":", 1)[1].isdigit():
used[name.split(":", 1)[1]] = 1
elif name.isdigit() and name in face_id_mapping:
used[name] = 1
for face_id in face_id_mapping.keys():
face_id = str(face_id)
if face_id.isdigit():
used[face_id] = 1
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():
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):
new_id_num = 1
if peoples:
id_nums = [int(p.name.split(':')[1]) for p in peoples if ':' in p.name and p.name.split(':')[1].isdigit()]
if id_nums: new_id_num = max(id_nums) + 1
else: new_id_num = len(peoples) + 1
new_id_num = allocate_next_face_id()
new_name = "ID:{}".format(new_id_num)
peoples.append(People(new_name, feature))
try: