# coding=gb2312 import json import os.path import sys from Database.manager_database import * from Speaker.speak_base import SpeakServer, beep from UWB.positioning_standalone_v2 import * FINISH_LINE_SIZE1 = 0 FINISH_LINE_SIZE2 = 1 RUNNING_COUNTING = "cnt" RUNNING_STATUS = "stat" RUNNING_MES = "mes" STATUS_ACROSS = "across" STATUS_EXIT = "exit" RUNNING_LAST_STATUS_TIME = "lst" RUNNING_COST = "cost" RUNNING_DONE = 'done' PATH = sys.path[0] ANCHOR_RSSI_ADJUSTMENT_PATH = f"{PATH}/Exercise3/anchor_adjustment.json" def score_compute(age, cost, score_data): my_age_index = 0 for age_index in range(len(score_data["age"])): candidate_age = score_data["age"][age_index] if candidate_age[0] <= age <= candidate_age[1]: my_age_index = age_index break my_score_index = len(score_data["values"]) - 1 for value_index in range(len(score_data["values"]) - 1, 0 - 1, -1): value = score_data["values"][value_index][my_age_index] if cost <= value[0] * 60 + value[1]: my_score_index = value_index else: break if cost > score_data["values"][-1][my_age_index][0] * 60 + score_data["values"][-1][my_age_index][1]: my_base_score = 0 else: my_base_score = score_data["score"][my_score_index] if my_base_score == 100: value = score_data["values"][my_score_index][my_age_index] base_score = value[0] * 60 + value[1] my_score = my_base_score + (base_score - cost) // 5 * 1 else: my_score = my_base_score return my_score class Running: def __init__(self, positioning: Positioning, round_num=3, min_round_time=30) -> None: self.min_round_time = float(min_round_time) self.round_num = int(round_num) self.positioning = positioning self.positioning_tag = None self.finish_line = {} # 起跑时间 self.start_time = 0 # 结束时间 self.stop_time = 0 # 跑步记录 self.running_record = {} # 跑步流水记录 self.running_record_list = {} # 跑步圈时记录 self.round_record = {} # 记录手环 self.test_tag = [] # 手环对应人员信息 self.person_mes = {} # 是否起跑状态 self.__start_running = threading.Event() self.__start_running.clear() # 说话 self.speak_driver = SpeakServer() self.speak_driver.start() # 成绩文件 file = open(f"{PATH}/Exercise3/running_score.json") self.score_data = json.load(file) # 基站校准数据 anchor_list = list(positioning.multi_uwb.uwb_driver_table.keys()) default_effective_rssi = -84 if os.path.exists(ANCHOR_RSSI_ADJUSTMENT_PATH): with open(ANCHOR_RSSI_ADJUSTMENT_PATH, "r", encoding="utf-8_sig") as file: self.anchor_rssi_adjustment = json.load(file) if len(self.anchor_rssi_adjustment) < len(anchor_list): self.anchor_rssi_adjustment = { anchor: default_effective_rssi for anchor in anchor_list } with open(ANCHOR_RSSI_ADJUSTMENT_PATH, "w", encoding="utf-8_sig") as file: json.dump(self.anchor_rssi_adjustment, file) else: self.anchor_rssi_adjustment = { anchor: default_effective_rssi for anchor in anchor_list } with open(ANCHOR_RSSI_ADJUSTMENT_PATH, "w", encoding="utf-8_sig") as file: json.dump(self.anchor_rssi_adjustment, file) # 跑步记录数据 self.final_data = {} # 是否播报 self.is_play = False def set_play(self, is_play): self.is_play = is_play def update_and_play(self, tag, new_record): if self.is_play: this_record = self.running_record[tag] if this_record[RUNNING_COUNTING] != new_record[RUNNING_COUNTING]: person_name = self.person_mes[tag][NAME] person_id = self.person_mes[tag][ID] if new_record[RUNNING_DONE]: self.speak_driver.add_speak(f"{person_name}完成考试！") else: self.speak_driver.add_speak(f"{person_name}通过第{new_record[RUNNING_COUNTING]}圈！") self.running_record[tag] = new_record def set_config(self, round_num, min_round_time): self.min_round_time = float(min_round_time) self.round_num = int(round_num) def add_tag(self, tag): self.test_tag.append(tag) def del_tag(self, tag): self.test_tag.remove(tag) del self.person_mes[tag] def add_tag_mes(self, tag, mes): self.person_mes[tag] = mes def reset(self): self.test_tag.clear() self.person_mes.clear() self.start_time = 0 def start(self): # 清除缓存 self.__start_running.clear() self.round_record.clear() self.positioning.clear_information() self.positioning.resume() # 初始化记录 self.running_record = { tag: { RUNNING_COUNTING: -1, RUNNING_STATUS: STATUS_EXIT, RUNNING_MES: "未起始", RUNNING_COST: 0, RUNNING_DONE: False, } for tag in self.test_tag } # 起跑播报 self.speak_driver.add_speak("准备开始考试，请就位，倒计时：") counting = 5 for i in range(counting): start_time = time.time() self.speak_driver.add_speak(counting - i) while time.time() - start_time < 1: time.sleep(0.05) self.speak_driver.wait_4_speak() self.start_time = time.time() self.positioning.set_valid_time(self.start_time) beep(duration=700, freq=640) self.__start_running.set() # 启动计算线程 threading.Thread(target=self._thread_running, daemon=True).start() threading.Thread(target=self._thread_processing, daemon=True).start() def kill(self): # 记录手环 self.test_tag.clear() self.running_record_list.clear() self.__start_running.clear() def stop(self): # # 起跑时间 self.stop_time = time.time() # self.start_time = 0 self.kill() # 计算最终成绩 self.final_data.clear() raw_score = self.get_score() for score in raw_score: band_id = score["band_id"] person_mes = self.person_mes[band_id] tag_mes = self.running_record[band_id] score.update(person_mes) score.update(tag_mes) person_id = score["id"] round_time = self.round_record.get(person_id) round_time_record = round_time if round_time else [] score.update({"round_time": round_time_record}) score.update({ "fix": False, "final_result": round_time_record }) self.final_data[person_id] = score # 获得当前所有成绩 def get_all_score(self): return list(self.final_data.values()) # 成绩修复 def fix_score(self, person_id): if not self.final_data[person_id][RUNNING_DONE]: detected_round = len(self.final_data[person_id]["round_time"]) lacked_round = self.round_num - detected_round + 1 result = [ self.final_data[person_id]["round_time"][i] for i in range(detected_round) ] # 如果没有成绩，按最终完成时间去计算 if len(result) <= 1: total_time = self.stop_time - self.start_time result = [total_time / (self.round_num + 1) for i in range(self.round_num + 1)] # 如果有成绩 else: for _ in range(lacked_round): max_result = max(result) index = result.index(max_result) result = result[0:index:] + [max_result / 2, max_result / 2] + result[index + 1::] self.final_data[person_id]["final_result"] = { i: result[i] for i in range(len(result)) } self.final_data[person_id]["fix"] = True self.final_data[person_id]["total_time"] = sum(result) # 修复撤销 def fix_withdraw(self, person_id): self.final_data[person_id]["final_result"] = self.final_data[person_id]["round_time"] self.final_data[person_id]["fix"] = False def _thread_processing(self): while self.__start_running.is_set(): data = self.positioning.get_last_information() try: # 过滤无效操作 if data: record_time, mes = data # print(self.start_time - record_time, len(self.positioning.tag_information)) # if record_time < self.start_time - 3: # continue tag = mes[TAG] if tag not in self.test_tag: continue distance = mes[DIST] rssi = mes[RSSI] anchor = mes[ANCHOR_ID] self.running_record_list.setdefault(tag, []) self.running_record_list[tag].append( {RECORD: record_time, DIST: distance, RSSI: rssi, ANCHOR_ID: anchor} ) except Exception as e: traceback.format_exc() print(f"获取数据时发生错误：{e.args}") def _thread_running(self): # 提前开始检测时间 bias = 3 # 判定周期时间 judge_window_time = 6 detect_time_range = judge_window_time / 2 # 有效进入距离 effective_distance = 300 # effective_rssi = -84 # 开始判定距离 judge_distance = 700 # 离开判定时间 leave_detect_time = 10 while self.__start_running.is_set(): this_time = time.time() for tag, record in self.running_record_list.copy().items(): if tag not in self.test_tag or self.running_record[tag][RUNNING_DONE]: continue # 获取测试人员id person_id = self.person_mes[tag]["id"] sorted_record = sorted(record, key=lambda x: x[RECORD]) this_tag = { RUNNING_COUNTING: -1, RUNNING_STATUS: STATUS_EXIT, RUNNING_MES: "离开", RUNNING_COST: 0, RUNNING_DONE: False } # Cache # 上次进入检测区域的时间 last_enter_time = self.start_time # 上次离开检测区域的时间 last_leave_time = -1 # 最后一条记录的时间 record_time = self.start_time # print("===============================================================================================") for one_record in sorted_record: # print(one_record) try: # 消除无效数据 if one_record[RECORD] < self.start_time - bias or this_time - one_record[RECORD] < detect_time_range: continue record_time = one_record[RECORD] distance = one_record[DIST] rssi = one_record[RSSI] anchor = one_record[ANCHOR_ID] # 更新时间窗口中的数据，未来时间一段时间内的数据。 frontward_window = list( filter( lambda x: x[DIST] is not None and -detect_time_range <= record_time - x[RECORD] < 0, sorted_record ) ) # 离开判定1 if ( # 距离上次进入冲线条件的时间已经有一段时间 record_time - last_leave_time > leave_detect_time and this_tag[RUNNING_STATUS] == STATUS_ACROSS ): this_tag[RUNNING_STATUS] = STATUS_EXIT this_tag[RUNNING_MES] = "离开" # print("离开判定") # 获取有效信号值 effective_rssi = self.anchor_rssi_adjustment[anchor] # 冲线判定 if ( ( # CASE1：如果未来距离没有数据 len(frontward_window) == 0 and ( # 并且检测到距离小于判定距离 (distance is not None and distance < judge_distance) # 或者检测到的信号强度强于检测强度 or rssi >= effective_rssi ) ) or ( # CASE2：如果未来还有数据 len(frontward_window) > 0 and distance is not None and distance <= effective_distance # 未来距离都比现在大 and distance - min([fw[DIST] for fw in frontward_window]) <= 20 ) ): # print("冲线判定") if ( # 并且当前时间距离上次已经大于单圈时间，并且已经离开了 record_time - last_leave_time > self.min_round_time and this_tag[RUNNING_STATUS] == STATUS_EXIT ): # print("圈数判定") this_tag[RUNNING_STATUS] = STATUS_ACROSS this_tag[RUNNING_MES] = "冲线" this_tag[RUNNING_COUNTING] += 1 # 圈数时间记录 self.round_record.setdefault(person_id, {}) this_round_cost = 0 if record_time - last_enter_time < 0 else record_time - last_enter_time self.round_record[person_id][this_tag[RUNNING_COUNTING]] = this_round_cost last_enter_time += this_round_cost # 结束判定 if this_tag[RUNNING_COUNTING] == self.round_num: this_tag[RUNNING_DONE] = True this_tag[RUNNING_COST] = record_time - self.start_time break last_leave_time = record_time # 如果还处于冲线状态，并且距离上次检测时间不超过离开判定周期，更新冲线时间 if ( this_tag[RUNNING_STATUS] == STATUS_ACROSS # and record_time - last_leave_time < detect_time_range ): last_leave_time = record_time except Exception as e: print(traceback.format_exc()) print(f"计算时发生错误:{e.args}") # 离开判定2 if ( # 最后离开的时间已经超过了离开判定时间 this_time - last_leave_time > leave_detect_time # 最后的一条数据距离现在已经超过了离开判定时间 and this_time - record_time > leave_detect_time and this_tag[RUNNING_STATUS] == STATUS_ACROSS ): this_tag[RUNNING_STATUS] = STATUS_EXIT this_tag[RUNNING_MES] = "离开" if this_time - last_leave_time > self.min_round_time and this_tag[RUNNING_STATUS] == STATUS_EXIT: this_tag[RUNNING_MES] = "回程" # print(this_tag) # 更新记录 self.update_and_play(tag=tag, new_record=this_tag) # 休息一下 time.sleep(0.5) def get_person_round_time(self, person_id): if person_id not in self.final_data.keys(): return {} return self.final_data[person_id]["final_result"] def get_valid_score(self): result = [] for person_id, data in self.final_data.items(): if data["done"] or data["fix"]: data["score"] = score_compute(data["age"], data["total_time"], self.score_data) result.append(data) return result # 获取成绩接口 def get_score(self): score = [] this_cost_time = time.time() - self.start_time # print(self.running_record) for tag, record in self.running_record.items(): if record[RUNNING_COUNTING] < 0: finish_status = "未起始" else: if record[RUNNING_COUNTING] == 0: finish_status = f"已开始({record[RUNNING_MES]})" else: finish_status = f"考试中({record[RUNNING_MES]})" if record[RUNNING_COUNTING] == self.round_num: finish_status = "已完成考试" age = self.person_mes[tag]["age"] one = { "band_id": tag, "score": score_compute(age, record[RUNNING_COST], self.score_data) if record[RUNNING_COUNTING] == self.round_num else 0, "total_time": record[RUNNING_COST] if record[RUNNING_DONE] or self.start_time == 0 else this_cost_time, "round": record[RUNNING_COUNTING] if record[RUNNING_COUNTING] > 0 else 0, "finish": finish_status, "percentage": record[RUNNING_COUNTING] / self.round_num * 100 if record[RUNNING_COUNTING] > 0 else 0 } score.append(one) # print(score) return score