415 lines
17 KiB
Python
415 lines
17 KiB
Python
# coding=gb2312
|
|
import json
|
|
import sys
|
|
|
|
from Database.manager_database import *
|
|
from Speaker.speak_base import SpeakServer, beep
|
|
from UWB.positioning_standalone 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]
|
|
|
|
|
|
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)
|
|
# 跑步记录数据
|
|
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()
|
|
|
|
def start(self):
|
|
# 清除缓存
|
|
self.__start_running.clear()
|
|
self.round_record.clear()
|
|
self.positioning.clear_information()
|
|
self.positioning.resume()
|
|
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)
|
|
duration = 700 # millisecond
|
|
freq = 640 # Hz
|
|
beep(freq, duration)
|
|
self.__start_running.set()
|
|
# 初始化记录
|
|
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
|
|
}
|
|
# 启动计算线程
|
|
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:
|
|
# print(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
|
|
avg_distance = mes[DIST]
|
|
avg_rssi = mes[RSSI]
|
|
self.running_record_list.setdefault(tag, [])
|
|
self.running_record_list[tag].append(
|
|
{RECORD: record_time, DIST: avg_distance, RSSI: avg_rssi}
|
|
)
|
|
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 = -80
|
|
# 开始判定距离
|
|
judge_distance = 700
|
|
judge_rssi = -82
|
|
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_status_time = -1
|
|
last_enter_time = self.start_time
|
|
for one_record in sorted_record:
|
|
try:
|
|
record_time = one_record[RECORD]
|
|
# 消除无效数据
|
|
if record_time < self.start_time - bias or this_time - record_time < detect_time_range:
|
|
continue
|
|
distance = one_record[DIST]
|
|
rssi = one_record[RSSI]
|
|
# 更新时间窗口中的数据
|
|
frontward_window = list(
|
|
filter(
|
|
lambda x: -detect_time_range <= record_time - x[RECORD] < 0,
|
|
sorted_record
|
|
)
|
|
)
|
|
# 冲线判定
|
|
if (
|
|
(
|
|
# CASE1 未来没有减少的趋势
|
|
len(frontward_window) == 0
|
|
# 未来距离都比现在大
|
|
or min([fw[DIST] for fw in frontward_window]) > distance
|
|
# 未来信号都比现在弱
|
|
or max([fw[RSSI] for fw in frontward_window]) < rssi
|
|
) or (
|
|
# CASE2 小于有效距离
|
|
distance <= effective_distance
|
|
or rssi >= effective_rssi
|
|
)
|
|
) and (
|
|
# 并且时间大于单圈时间，且已经离开，并且达到判定距离
|
|
record_time - last_status_time > self.min_round_time
|
|
and this_tag[RUNNING_STATUS] == STATUS_EXIT
|
|
and (distance < judge_distance or rssi >= judge_rssi)
|
|
):
|
|
last_status_time = record_time
|
|
this_tag[RUNNING_STATUS] = STATUS_ACROSS
|
|
this_tag[RUNNING_MES] = "冲线"
|
|
this_tag[RUNNING_COUNTING] += 1
|
|
# 圈数时间记录
|
|
self.round_record.setdefault(person_id, {})
|
|
self.round_record[person_id][
|
|
this_tag[RUNNING_COUNTING]
|
|
] = 0 if record_time - last_enter_time < 0 else record_time - last_enter_time
|
|
last_enter_time += self.round_record[person_id][
|
|
this_tag[RUNNING_COUNTING]
|
|
]
|
|
# 结束判定
|
|
if this_tag[RUNNING_COUNTING] == self.round_num:
|
|
this_tag[RUNNING_DONE] = True
|
|
this_tag[RUNNING_COST] = record_time - self.start_time
|
|
|
|
# 离开判定1
|
|
elif (
|
|
(
|
|
# 如果还有数据，并且之前已经进入了判定距离
|
|
len(frontward_window) > 0
|
|
and
|
|
(
|
|
min([fw[DIST] for fw in frontward_window]) > judge_distance
|
|
or max([fw[RSSI] for fw in frontward_window]) < judge_rssi
|
|
)
|
|
) and (
|
|
# 并且当前距离已离开范围，当前状态为冲线，距离上次判定时间超过一个判定间隔
|
|
(
|
|
distance > effective_distance
|
|
or rssi < effective_rssi
|
|
)
|
|
and this_tag[RUNNING_STATUS] == STATUS_ACROSS
|
|
and record_time - last_status_time > detect_time_range
|
|
)
|
|
):
|
|
last_status_time = record_time
|
|
this_tag[RUNNING_STATUS] = STATUS_EXIT
|
|
this_tag[RUNNING_MES] = "离开"
|
|
|
|
# 离开判定2
|
|
if (
|
|
(
|
|
# 如果已无数据
|
|
len(frontward_window) == 0
|
|
) and (
|
|
# 并且当前距离已离开范围，当前状态为冲线
|
|
(distance > 0 or rssi < -60)
|
|
# 或者最后一条数据已经过了离开判定时间了
|
|
or this_time - record_time > detect_time_range
|
|
)
|
|
and this_tag[RUNNING_STATUS] == STATUS_ACROSS
|
|
):
|
|
last_status_time = record_time
|
|
this_tag[RUNNING_STATUS] = STATUS_EXIT
|
|
this_tag[RUNNING_MES] = "离开"
|
|
except Exception as e:
|
|
traceback.format_exc()
|
|
print(f"计算时发生错误:{e.args}")
|
|
|
|
if this_time - last_status_time > self.min_round_time and this_tag[RUNNING_STATUS] == STATUS_EXIT:
|
|
this_tag[RUNNING_MES] = "回程"
|
|
# 更新记录
|
|
self.update_and_play(tag=tag, new_record=this_tag)
|
|
# 休息一下
|
|
time.sleep(1)
|
|
|
|
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
|
|
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] 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
|
|
if record[RUNNING_COUNTING] > 0 else 0
|
|
}
|
|
score.append(one)
|
|
return score
|