# -*- coding: utf-8 -*- import base64 import copy import json import re import shutil import threading from copy import deepcopy from typing import List import requests from flask import jsonify, request, Response, send_from_directory, send_file import concurrent.futures import queue import LSZXBackend.base_driver from Backend.consensus import * # from Database.manager_database import * from DeviceDefine.consensus import UNKNOWN, MASTER, SLAVER from Exercise3.base_exercise import base_detect_image from Exercise3.overhang import Overhang from Exercise3.tricep_dip import Tricep_dip_1 from Exercise3.pull_up_1 import PullUp_1 from Exercise3.pull_up_2 import PullUp_2 from Exercise3.pull_up_3 import PullUp_3 from Exercise3.push_up_1 import PushUp_1 from Exercise3.push_up_2 import PushUp_2 from Exercise3.push_up_3 import PushUp_3 from Exercise3.run_around import Runaround # 调试 # from Exercise3.running_muwb import Running from Exercise3.running_muwb_with_rssi import Running import Exercise3.running_muwb_pure_detection from Exercise3.sit_up_new import SitUp_New from Exercise3.sit_up_old_1 import SitUp_Old_1 from Exercise3.sit_up_old_2 import SitUp_Old_2 from Exercise3.sit_up_old_3 import SitUp_Old_3 from LSZXBackend.eng_code import FUNC_TABLE, FACTORY_MODE from LSZXBackend.general import * from LSZXVideo.video_recording import VideoRecording from LSZXVideo.video_sender import VideoSending from Other.np_encoder import NpEncoder from Speaker.fake_speak_base import beep from Database.database_mgr import * # 难度值 DIFFICULT_HARD = 1 DIFFICULT_EASY = 2 STANDARDS_NEW = 3 STANDARDS_OLD = 4 COMMON_EASY = 5 # 训练难度 difficulty_lookup = { DIFFICULT_HARD: "比武难度", DIFFICULT_EASY: "考核难度", COMMON_EASY: "普通难度" } # 动作标准 standards_lookup = { STANDARDS_NEW: "新式动作标准", STANDARDS_OLD: "旧式动作标准" } DIFFICULT_FILE = "./train_info.json" SUMMARY_FILE = "./summary_info.json" DIFFICULTY_MES = "difficulty_mes" DIFFICULTY_DIS = "difficulty_display" DIFFICULTY = "difficulty" STANDARDS_DIS = "standards_display" STANDARDS = "standards" # 状态值 MAX_HR = "max_hr" MIN_BO = "min_bo" HR = "hr" BO = "bo" # 缓存路径 CACHE_DIR = "./Cache/" STARTING = 1 STOP = 0 # 准备录屏 video_path = os.path.join(GLOBAL_DIR, "LSZXVideo", "Video") output_video_path = os.path.join(GLOBAL_DIR, "LSZXVideo", "Output_Video") video_recorder = VideoRecording() video_recorder.start() # 视频发送 video_sending = VideoSending() # 用于在线程池中发送POST请求 def send_post_request(q, url_target): try: while not q.empty(): seg_data = q.get() _response = requests.post(url=url_target, json=seg_data, timeout=3) if _response.status_code != 200: print(f"请求网络：{url_target} 响应不成功") q.task_done() except requests.RequestException: print('转发失败！！！') print(traceback.format_exc()) class ExerciseBackend(LSZXBackend.base_driver.BaseDriver): def __init__( self, name, domain="0.0.0.0", port=0, master_mode=True, positioning=True, camera=True, speaker=True, multi_positioning_mode=True, device_type=UNKNOWN, **kwargs ): super().__init__( name, domain, port, master_mode, positioning=positioning, camera=camera, speaker=speaker, multi_positioning_mode=multi_positioning_mode, device_type=device_type, **kwargs ) # 管理难度 # 生成初始化动作难度与标准json文件 if not os.path.exists(DIFFICULT_FILE): self.train_info = { DIFFICULTY_DIS: difficulty_lookup[DIFFICULT_HARD], DIFFICULTY: DIFFICULT_HARD, STANDARDS_DIS: standards_lookup[STANDARDS_OLD], STANDARDS: STANDARDS_OLD, } try: with open(DIFFICULT_FILE, "w+", encoding='utf-8_sig') as f: json.dump(self.train_info, f, ensure_ascii=False) except: pass else: try: with open(DIFFICULT_FILE, "r", encoding='utf-8_sig') as f: self.train_info = json.load(f) except: pass # 手环状态监控 self.hr_bo_record = {MAX_HR: 0, MIN_BO: 100, HR: 0, BO: 100} # 同步状态监控 self.synchronization_kill_sign = True self.synchronization_message = { TASK_HAD_DONE: 0, STATUS: STOP, TASK_NUMBER: 0 } # 当前作训工程 self.project = None self.exercise_tag = "" self.running = None self.running_mes = {} # 数据同步缓存 self.synchronization_cache = [] # 实时展示 # 初始化传送管理端的成绩 self.synchronization_info = {} self.statistics_score = {SITUP: [], RUNAROUND: [], PUSHUP: [], PULLUP: [], RUNNING: [], 'total': [], 'top': {'total': {}, SITUP: {}, RUNAROUND: {}, PULLUP: {}, PUSHUP: {}, RUNNING: {}}} self._statistics_score = {} self._statistics_sql_score = {} self._sql_score = {} self.hrbo_info = {} self._hrbo_info = {} self.closed_chamber = {} self.HrBo_info = {} self.tag_mes = {HR: None, BO: None} if device_type == MASTER or device_type == SLAVER: self.send_score_signal = threading.Event() self.send_score_signal.set() self.send_sql_score_signal = threading.Event() # self.send_sql_score_signal.set() threading.Thread(target=self.send_score).start() threading.Thread(target=self.send_sql_score).start() else: self.manager.delete_orphan_batches() self.manager.delete_empty_score_type() self.sql_data = self.manager.select_all_score() for i in self.sql_data: i.update({'hr': {'hr': '', 'color': 'W'}, 'bo': {'bo': '', 'color': 'W'}}) self._sql_score.update({i['id']: copy.deepcopy(i)}) for key in i.keys(): if '_count' in key: if i[key]: score = str(i[key]) else: score = '' i[key] = {'count': score, 'is_changed': False} threading.Thread(target=self.processing_sql).start() self.processing_score_signal = threading.Event() threading.Thread(target=self.processing_score).start() # 汇总分析 if not os.path.exists(SUMMARY_FILE): batch = self.manager.get_this_batch() if not batch: this_time = int(time.time()) batch = time.strftime("%Y年%m月%d日%H时", time.localtime(this_time)) self.summary_info = { 'batch': batch, 'default_projects': {SITUP: '仰卧起坐', RUNAROUND: '30*2蛇形跑', PULLUP: '引体向上', RUNNING: '长跑'}, 'default_batch': batch, 'default_class_projects': {RUNNING: '长跑'}, 'default_class_batch': batch, 'default_group_projects': {RUNNING: '长跑'}, 'default_team_project': {RUNNING: '长跑'}, 'default_team_batch': batch, 'default_team_class': ['1班'], 'default_personal_projects': {SITUP: '仰卧起坐', RUNAROUND: '30*2蛇形跑', PULLUP: '引体向上', RUNNING: '长跑'}, 'default_personal_batch': [batch], 'default_personal_personal': {'id': 1, 'name': '人员1'} } try: with open(SUMMARY_FILE, "w+", encoding='utf-8_sig') as f: json.dump(self.summary_info, f, ensure_ascii=False) except: pass else: try: with open(SUMMARY_FILE, "r", encoding='utf-8_sig') as f: self.summary_info = json.load(f) batch = self.manager.get_this_batch() if batch: self.summary_info['batch'] = self.manager.get_this_batch() except: batch = self.manager.get_this_batch() if not batch: this_time = int(time.time()) batch = time.strftime("%Y年%m月%d日%H时", time.localtime(this_time)) self.summary_info = { 'batch': batch, 'default_projects': {SITUP: '仰卧起坐', RUNAROUND: '30*2蛇形跑', PULLUP: '引体向上', RUNNING: '长跑'}, 'default_batch': batch, 'default_class_projects': {RUNNING: '长跑'}, 'default_class_batch': batch, 'default_group_projects': {RUNNING: '长跑'}, 'default_team_project': {RUNNING: '长跑'}, 'default_team_batch': batch, 'default_team_class': ['1班'], 'default_personal_projects': {SITUP: '仰卧起坐', RUNAROUND: '30*2蛇形跑', PULLUP: '引体向上', RUNNING: '长跑'}, 'default_personal_batch': [batch], 'default_personal_personal': {'id': 1, 'name': '人员1'} } # 更新汇总信息json def _update_summary_info(self): try: with open(SUMMARY_FILE, "w", encoding='utf-8_sig') as f: json.dump(self.summary_info, f, ensure_ascii=False) except: print(traceback.format_exc()) def reset_hr_bo_record(self): self.hr_bo_record = {MAX_HR: 0, MIN_BO: 100, HR: 0, BO: 100} def update_hr_bo_record(self, tag_mes): if tag_mes[HR] != 0: self.hr_bo_record[HR] = tag_mes[HR] if self.hr_bo_record[MAX_HR] < tag_mes[HR]: self.hr_bo_record[MAX_HR] = tag_mes[HR] if tag_mes[BO] != 0: self.hr_bo_record[BO] = tag_mes[BO] if self.hr_bo_record[MIN_BO] > tag_mes[BO]: self.hr_bo_record[MIN_BO] = tag_mes[BO] def _get_train_info(self): try: with open(DIFFICULT_FILE, 'r', encoding='utf-8_sig') as read_json: self.train_info = json.load(read_json) difficulty = self.train_info[DIFFICULTY] standards = self.train_info[STANDARDS] return difficulty, standards except: if os.path.exists(DIFFICULT_FILE): os.remove(DIFFICULT_FILE) self.train_info = { DIFFICULTY_DIS: difficulty_lookup[DIFFICULT_HARD], DIFFICULTY: DIFFICULT_HARD, STANDARDS_DIS: standards_lookup[STANDARDS_OLD], STANDARDS: STANDARDS_OLD, } with open(DIFFICULT_FILE, "w+", encoding='utf-8_sig') as f: json.dump(self.train_info, f, ensure_ascii=False) difficulty = self.train_info[DIFFICULTY] standards = self.train_info[STANDARDS] return difficulty, standards def _set_train_info(self, train_info): try: difficulty = train_info.get(DIFFICULTY) standards = train_info.get(STANDARDS) if not difficulty or not standards: return self.train_info = { DIFFICULTY_DIS: difficulty_lookup[difficulty], DIFFICULTY: difficulty, STANDARDS_DIS: standards_lookup[standards], STANDARDS: standards } with open(DIFFICULT_FILE, "w+", encoding='utf-8_sig') as f: json.dump(self.train_info, f, ensure_ascii=False) except: print(traceback.format_exc()) def get_band_mes(self): data = request.json try: band_id = data.get(BAND_ID) if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] return jsonify({STATUS: STATUS_OK, DATA: info}) else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 准备进行俯卧撑 def active_push_up(self): if self.project: self.project.kill() data = request.json person_id = data.get(ID) band_id = data.get(BAND_ID) if person_id: info = self.manager.select_a_person(person_id)[0] else: if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) try: difficulty, standards = self._get_train_info() if difficulty == DIFFICULT_HARD: self.project = PushUp_1(info, camera=self.camera) elif difficulty == DIFFICULT_EASY: self.project = PushUp_2(info, camera=self.camera) elif difficulty == COMMON_EASY: self.project = PushUp_3(info, camera=self.camera) self.exercise_tag = PUSHUP return jsonify({STATUS: STATUS_OK}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 准备进行引体向上 def active_pull_up(self): if self.project: self.project.kill() data = request.json person_id = data.get(ID) band_id = data.get(BAND_ID) if person_id: info = self.manager.select_a_person(person_id)[0] else: if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) try: difficulty, standards = self._get_train_info() if difficulty == DIFFICULT_HARD: self.project = PullUp_1(info, camera=self.camera) elif difficulty == DIFFICULT_EASY: self.project = PullUp_2(info, camera=self.camera) elif difficulty == COMMON_EASY: self.project = PullUp_3(info, camera=self.camera) self.exercise_tag = PULLUP return jsonify({STATUS: STATUS_OK}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 准备进行曲臂悬垂 def active_overhang(self): if self.project: self.project.kill() data = request.json person_id = data.get(ID) band_id = data.get(BAND_ID) if person_id: info = self.manager.select_a_person(person_id)[0] else: if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) try: self.project = Overhang(info, camera=self.camera) self.exercise_tag = OVERHANG return jsonify({STATUS: STATUS_OK}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 准备进行双杠臂屈伸 def active_tricep_dip(self): if self.project: self.project.kill() data = request.json person_id = data.get(ID) band_id = data.get(BAND_ID) if person_id: info = self.manager.select_a_person(person_id)[0] else: if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) try: self.project = Tricep_dip_1(info, camera=self.camera) self.exercise_tag = TRICEDIP return jsonify({STATUS: STATUS_OK}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 准备进行仰卧起坐 def active_sit_up(self): if self.project: self.project.kill() data = request.json person_id = data.get(ID) band_id = data.get(BAND_ID) if person_id: info = self.manager.select_a_person(person_id)[0] else: if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) try: difficulty, standards = self._get_train_info() if standards == STANDARDS_NEW: self.project = SitUp_New(info, camera=self.camera) elif standards == STANDARDS_OLD and difficulty == DIFFICULT_HARD: self.project = SitUp_Old_1(info, camera=self.camera) elif standards == STANDARDS_OLD and difficulty == DIFFICULT_EASY: self.project = SitUp_Old_2(info, camera=self.camera) elif standards == STANDARDS_OLD and difficulty == COMMON_EASY: self.project = SitUp_Old_3(info, camera=self.camera) self.exercise_tag = SITUP return jsonify({STATUS: STATUS_OK}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 准备进行蛇形跑 def active_run_around(self): if self.project: self.project.kill() data = request.json person_id = data.get(ID) band_id = data.get(BAND_ID) if person_id: info = self.manager.select_a_person(person_id)[0] else: if band_id: raw_info = self.manager.select_a_person_from_band(band_id) if not raw_info: return jsonify({STATUS: ID_ERROR, MSG: "该手环没有匹配的人员"}) info = raw_info[0] else: return jsonify({STATUS: ILLEGAL_CONFIG, MSG: "参数缺失"}) try: self.project = Runaround(info, camera=self.camera) self.exercise_tag = RUNAROUND return jsonify({STATUS: STATUS_OK}) except Exception as e: return jsonify({STATUS: ACTION_FAIL, MSG: str(e)}) # 开始进行作训 def start_exercise(self): try: self.reset_hr_bo_record() # self.positioning.pause() self.camera.start_record() video_name = self.project.info[ID] + '_' + self.project.info[ "name"] + '_' + self.project.exercise_type + '_' + 'unknow' # 开始屏幕录制 video_recorder.write_start(video_name) self.project.start() self.project.waiting_for_start() # 清空传送管理端的成绩 self.synchronization_info = {} self.tag_mes = {HR: None, BO: None} return jsonify({STATUS: STATUS_OK, "valid": self.project.is_valid(), "mes": self.project.valid_mes()}) except Exception as e: print(traceback.format_exc()) error_msg = { STATUS: STATUS_UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e) } return jsonify(error_msg) # 强制结束作训测试 def stop_exercise(self): try: if self.project: info = self.project.get_info() else: info = None self.camera.stop_record() self.camera.clear_cache() self.project.kill() result = self.project.get_result() score = result["score"] if score == '不合格': real_score = 0 else: real_score = float(score) video_name = self.project.info[ID] + '_' + self.project.info[ "name"] + '_' + self.project.exercise_type + '_' + str(real_score) # 结束屏幕录制 # video_recorder.write_stop(video_name) self.synchronization_info[self.exercise_tag] = {} self.send_score_signal.set() # self.send_sql_score_signal.set() return jsonify({STATUS: STATUS_OK, "info": info}) except Exception as e: print(traceback.format_exc()) error_msg = { STATUS: STATUS_UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e) } return jsonify(error_msg) # 直接获取骨架信息 def video_directly(self): video_response = Response(base_detect_image(self.camera), mimetype='multipart/x-mixed-replace; boundary=frame') return video_response # 获得作训摄像头流 def exercise_video(self): if self.project is None: video_response = Response(base_detect_image(self.camera), mimetype='multipart/x-mixed-replace; boundary=frame') elif self.project and not self.project.is_start: video_response = Response(self.project.video_gen(), mimetype='multipart/x-mixed-replace; boundary=frame') else: video_response = Response(self.project.streaming_gen(), mimetype='multipart/x-mixed-replace; boundary=frame') return video_response # 录入作训成绩 def update_score(self): try: person_id = self.project.info[ID] name = self.project.info[NAME] person_class = self.project.info['class'] result = self.project.get_result() score = result["score"] had_done = result['had_done'] if score == '不合格': real_score = 0 else: real_score = float(score) count = result["count"] response_code = OK self.synchronization_info[self.exercise_tag] = {} self.send_score_signal.set() self._statistics_sql_score = { self.exercise_tag: {person_id: {'record': count, 'score': real_score}}} if had_done: self.send_sql_score_signal.set() _count = count _score = real_score elif self.exercise_tag in {PULLUP, TRICEDIP} and had_done == False: # 结束屏幕录制 video_name = self.project.info[ID] + '_' + self.project.info[ "name"] + '_' + self.project.exercise_type + '_' + str(real_score) video_recorder.write_stop(video_name) _count = count _score = real_score else: _count = -1 _score = 0 _result = self.manager.select_a_score(person_id=person_id, score_type=self.exercise_tag) if _result is None: response_code = self.manager.insert_a_score(record=_count, score=_score, person_id=person_id, score_type=self.exercise_tag, person_class=person_class, name=name) elif _count != -1: _record = float(_result[0]['record']) _score = float(_result[0]['score']) if self.exercise_tag in {PUSHUP, PULLUP, SITUP, OVERHANG, TRICEDIP}: if float(count) > _record or real_score > _score: response_code = self.manager.update_counting_score(record=count, score=real_score, person_id=person_id, score_type=self.exercise_tag) print(response_code) elif self.exercise_tag == RUNAROUND: if count != -1 and (float(count) < _record or real_score > _score): response_code = self.manager.update_timekeeping_score(record=_count, score=real_score, person_id=person_id, score_type=self.exercise_tag, ) else: response_code = STATUS_UNKNOWN_ERROR return_mes = "" except Exception as e: return_mes = str(e) response_code = STATUS_UNKNOWN_ERROR print(traceback.format_exc()) return jsonify({STATUS: response_code, "mes": return_mes}) # 获取作训成绩Websocket接口 def get_score(self, ws): while True: try: # 每隔0.3秒发送一次当前成绩 time.sleep(0.3) score_result = self.project.get_result() band_id = self.project.info.get("band_id") tag_mes = self.positioning.get_tag_mes(band_id) temp = {} if tag_mes: self.update_hr_bo_record(tag_mes) temp['hr'] = tag_mes[HR] temp['bo'] = "{:.2%}".format(tag_mes[BO] / 100) if tag_mes[HR] >= 190 or tag_mes[BO] <= 80: temp['normal'] = False else: temp['normal'] = True self.tag_mes = {HR: temp['hr'], BO: temp['bo']} else: if self.tag_mes[HR] and self.tag_mes[BO]: tag_mes = self.tag_mes else: tag_mes = {HR: '-', BO: '-'} temp['hr'] = tag_mes[HR] temp['bo'] = tag_mes[BO] temp['normal'] = True score_result.update(self.hr_bo_record) format_data = json.dumps(score_result, cls=NpEncoder) ws.send(format_data) temp['name'] = self.project.info[NAME] temp['id'] = self.project.info[ID] temp['count'] = score_result['count'] self.synchronization_info[self.exercise_tag] = temp self.send_score_signal.set() except Exception as e: # print(traceback.format_exc()) continue # 获得指定班级的数据 def get_class_data(self): try: data = request.json response_data = self.manager.get_a_class(data.get(CLASS)) return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得所有班级列表 def get_all_class(self): try: response_data = self.manager.get_all_class() return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得所有人员信息 def get_all_person(self): try: response_data = self.manager.select_all_person() return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得指定班级成绩 def get_a_class_score(self): try: data = request.json response_data = self.manager.get_a_class_score(data.get(CLASS)) return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得所有人员信息成绩 def get_all_score(self): try: response_data = self.manager.select_all_score() return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 開啓多基站長跑模式 def running_master_multi(self): self.running_mes.clear() try: self.running = Running(positioning=self.positioning) coordinates_data = { "master": [ [0, 0, "主机"] for _ in range(len(self.positioning.multi_uwb.uwb_driver_table)) ], "slaver": [] } self.speak_driver.add_speak("基站识别成功，请检查基站数量！") return jsonify({"status": OK, "coordinates": coordinates_data}) except Exception as e: print(traceback.format_exc()) error_msg = {"status": UNKNOWN_ERROR, "error_message": "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 開啓多基站長跑模式 def running_master_pd_mode(self): self.running_mes.clear() try: self.running = Exercise3.running_muwb_pure_detection.Running(positioning=self.positioning) coordinates_data = { "master": [ [0, 0, "主机"] for _ in range(len(self.positioning.multi_uwb.uwb_driver_table)) ], "slaver": [] } self.speak_driver.add_speak("基站识别成功，请检查基站数量！") return jsonify({"status": OK, "coordinates": coordinates_data}) except Exception as e: print(traceback.format_exc()) error_msg = {"status": UNKNOWN_ERROR, "error_message": "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 开启长跑主机模式 def running_master(self): self.running_mes.clear() # 检查网络链接 try: if not self.connection.get_connected_wifi() and not self.connection.master_mode: self.speak_driver.add_speak("网络未连接！") error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络未链接！"} return jsonify(error_msg) except Exception as e: print(traceback.format_exc()) error_msg = { STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络错误！", MSG: f"{e.args}" } self.speak_driver.add_speak("网络错误！请重新链接WiFi！") return jsonify(error_msg) if self.running: self.running.kill() # 重启驱动 self.positioning.resume() self.positioning.stop_positioning_mode() ip_list = [self.connection.get_self_ip()] + self.connection.get_other() if ip_list[0] is None: self.connection.wifi_ap.reboot() time.sleep(5) try: self.positioning.set_2_positioning_mode(10, ip_list) self.running = Running(positioning=self.positioning) anchor_coordinates = self.positioning.get_anchor_coordinates() if not anchor_coordinates: self.speak_driver.add_speak("定位出错，请重试或检查设备网络链接！") return jsonify({STATUS: ACTION_FAIL, ERROR_MESSAGE: "基站定位出错，请重试！"}) coordinates_data = { "master": [], "slaver": [] } master_anchor = self.running.get_master_anchor() for anchor, coordinate in anchor_coordinates.items(): if anchor == master_anchor: coordinates_data["master"].append([*coordinate, "主机"]) else: coordinates_data["slaver"].append([*coordinate, "从机"]) self.speak_driver.add_speak("基站识别成功，请检查基站数量！") return jsonify({STATUS: OK, "coordinates": coordinates_data}) except Exception as e: # self.positioning.pause() error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置长跑的圈时圈数 def set_running_config(self): data = request.json round_num = data.get("round") min_round_time = data.get("min_round_time") try: self.running.set_config(round_num, min_round_time) return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 重设长跑数据 def running_reset(self): try: # self.positioning.pause() self.running_mes = {} self.running.reset() return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 长跑增加一个手环 def running_add_band(self): data = request.json ids = data.get("id") try: self.running_mes = {} self.running.reset() test_person_name = [] for person_id in ids: person_mes = self.manager.select_a_person(person_id) band_id = person_mes[0]["band_id"] if not band_id: if person_mes: return jsonify({STATUS: ID_ERROR, ERROR_MESSAGE: f"{person_mes[0]['name']}未链接到人员信息！"}) else: return jsonify({STATUS: ID_ERROR, ERROR_MESSAGE: f"找不到人员编号为<{person_id}>的信息！"}) self.running_mes[band_id] = person_mes[0] self.running.add_tag(band_id) self.running.add_tag_mes(band_id, person_mes[0]) test_person_name.append(person_mes[0]["name"]) return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 长跑删除一个手环 def running_del_band(self): data = request.json band_id = data.get("band_id") try: del self.running_mes[band_id] self.running.del_tag(band_id) self.speak_driver.add_speak("删除成功!") return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 查看当前指定人员的圈时 def get_round_time(self): data = request.json person_id = data.get("id") try: data = self.running.get_person_round_time(person_id) result = [] time_counting = 0 if data: for key, value in data.items(): time_counting += value round_time = str(time.strftime("%H:%M:%S", time.gmtime(value))) counting_round_time = str(time.strftime("%H:%M:%S", time.gmtime(time_counting))) result.append({ "name": f"第{key}圈用时", "time": f"{round_time}({counting_round_time})", "no": key }) else: result = [] result = sorted(result, key=lambda x: x["no"]) return jsonify({STATUS: OK, DATA: result}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得所有考生详情 def get_running_all_score(self): try: data = {STATUS: OK, DATA: self.running.get_all_score()} return jsonify(data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 修正指定考生成绩 def fix_score(self): data = request.json person_id = data.get("id") try: self.running.fix_score(person_id) return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 撤销修正指定考生成绩 def fix_withdraw(self): data = request.json person_id = data.get("id") try: self.running.fix_withdraw(person_id) return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 长跑手环列表 def running_list(self): try: return jsonify(list(self.running_mes.values())) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 开始长跑 def start_running(self): try: self.running.start() return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置是否进行语音播报 def voice_play(self): _data = request.json is_play = _data.get("data") try: self.running.set_play(is_play) return jsonify({STATUS: OK, MSG: "设置成功"}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, MSG: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 结束长跑 def stop_running(self): try: # self.positioning.pause() self.synchronization_info[RUNNING] = [] self.send_score_signal.set() # self.send_sql_score_signal.set() if self.running: self.running.stop() return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 结束长跑 def stop_running_auto_clear(self): try: # self.positioning.pause() if self.running: self.running.stop() for person_mes in self.running_mes.values(): person_id = person_mes["id"] self.running.fix_score(person_id) return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得当前成绩 # 获取跑步成绩Websocket接口 def running_score(self, ws): while True: # 每隔0.2秒发送一次当前成绩 time.sleep(0.5) raw_score = self.running.get_score() final_score = [] # 对初始成绩进行排序 if raw_score: raw_score = sorted(raw_score, key=lambda x: (-x["score"], -x["round"], x["band_id"], ["total_time"])) rank = 0 last_total_time = -1 for score in raw_score: if score["total_time"] != last_total_time: rank += 1 last_total_time = score["total_time"] full_mes = deepcopy(score) band_id = score["band_id"] tag_mes = self.positioning.get_tag_status(band_id) person_mes = self.running_mes[band_id] full_mes.update(person_mes) full_mes.update({ "rank": rank }) full_mes.update(tag_mes) final_score.append(full_mes) temp = [] for row in final_score: if row[HR] != 0 and row[BO] != 0: hr = row[HR] bo = "{:.2%}".format(row[BO] / 100) if row[HR] >= 220 or row[BO] < 80: normal = False else: normal = True else: hr = '-' bo = '-' normal = True temp.append( {'id': row['id'], 'name': row['name'], 'hr': hr, 'bo': bo, 'rank': row['rank'], 'finish': row['finish'], 'round': row['round'], 'count': row['total_time'], 'normal': normal}) self.synchronization_info[RUNNING] = temp self.send_score_signal.set() format_data = json.dumps(final_score) ws.send(format_data) # 确认长跑成绩有效 def update_running_score(self): try: final_score = self.running.get_valid_score() self.synchronization_info[RUNNING] = [] self.send_score_signal.set() self._statistics_sql_score = {RUNNING: {}} for score in final_score: result = self.manager.select_a_score(person_id=score[ID], score_type=RUNNING) if result is None: self.manager.insert_a_score(record=score["total_time"], score=score['score'], person_id=score[ID], score_type=RUNNING, person_class=score['class'], name=score[NAME]) else: _record = float(result[0]['record']) _score = float(result[0]['score']) if float(score["total_time"]) < _record or float(score['score']) > _score: self.manager.update_timekeeping_score(record=score["total_time"], score=score['score'], person_id=score[ID], score_type=RUNNING, ) self._statistics_sql_score[RUNNING][score[ID]] = {'record': score["total_time"], 'score': score['score']} if final_score: self.send_sql_score_signal.set() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 修改考核难度和动作标准信息并保存 def set_train_info(self): try: data = request.json self._set_train_info(data) return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e.args)} return jsonify(error_msg) # 查看考核难度和动作标准信息 def view_train_info(self): try: result = copy.copy(self.train_info) result.update({STATUS: OK}) return jsonify(result) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 开启长跑从机模式 def running_slaver(self): error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "从机未开放长跑功能！"} self.speak_driver.add_speak("从机未开放长跑功能！") return jsonify(error_msg) # # 检查网络链接 # try: # if not self.connection.wifi.wifi_connect_status(): # error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络未链接！"} # return jsonify(error_msg) # except Exception as e: # error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络错误！", MSG: f"{e.args}"} # return jsonify(error_msg) # try: # return jsonify({STATUS: OK}) # except Exception as e: # error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} # return jsonify(error_msg) # 开始长跑 def start_running_slaver(self): try: self.positioning.resume() self.positioning.stop_positioning_mode() return jsonify({STATUS: OK}) except Exception as e: # self.positioning.pause() error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 结束长跑 def stop_running_slaver(self): try: # self.positioning.pause() self.positioning.stop_positioning_mode() return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 增加一个人员 def add_person(self): try: data = request.json response_code = self.manager.insert_a_info(data) # if response_code == OK: # self.manager.insert_a_score(data) return jsonify({STATUS: response_code}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 导入excel def load_xlsx(self): try: data = request.files if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR) file = data.get("file") save_file_dir = CACHE_DIR + "person.xlsx" file.save(CACHE_DIR + "person.xlsx") response_code = self.manager.insert_many_person_from_xlsx(save_file_dir) return jsonify({STATUS: response_code}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 更新某个人的数据 def update_person(self): try: data = request.json response_code = self.manager.update_a_person(data) return jsonify({STATUS: response_code}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 删除某个人的数据 def delete_person(self): try: data = request.json person_to_be_del = data.get(ID) if isinstance(person_to_be_del, List): for person in person_to_be_del: response_code = self.manager.delete_a_person(person) if response_code == OK: self.manager.delete_a_score(person) else: response_code = self.manager.delete_a_person(data.get(ID)) if response_code == OK: self.manager.delete_a_score(data.get(ID)) return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 删除所有人员信息 def delete_all_person(self): try: response_code = self.manager.delete_all_person() return jsonify({STATUS: response_code}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 导入成绩excel def load_score_xlsx(self): try: data = request.files if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR) file = data.get("file") save_file_dir = CACHE_DIR + "score.xlsx" file.save(CACHE_DIR + "score.xlsx") response_code = self.manager.insert_many_score_from_xlsx(save_file_dir) return jsonify({STATUS: response_code}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 搜索人员成绩信息 def get_a_name_score(self): try: data = request.json response_data = self.manager.select_a_score_from_name(data.get(NAME)) return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 搜索人员基本信息 def get_a_name_person(self): try: data = request.json response_data = self.manager.select_a_person_from_name(data.get(NAME)) return jsonify(response_data) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 导出所有人员信息成绩 def get_score_xlsx(self): try: self.manager.dump_score(CACHE_DIR + "score.xlsx") return send_from_directory(CACHE_DIR, "score.xlsx") except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 导出所有人员信息 def get_person_xlsx(self): try: self.manager.dump_person(CACHE_DIR + "person.xlsx") return send_from_directory(CACHE_DIR, "person.xlsx") except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 删除所有人员信息 def delete_all_score(self): try: response_code = self.manager.clear_all_score() return jsonify({STATUS: response_code}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 重置同步信息 def reset_synchronization_status(self): self.synchronization_message = { TASK_HAD_DONE: 0, STATUS: 0, TASK_NUMBER: 0 } def synchronization_server(self): try: cmd = request.json # print(cmd) if self.synchronization_message[STATUS] == STARTING: if cmd[MSG] == "start": self.synchronization_message[TASK_HAD_DONE] = 0 self.synchronization_message[TASK_NUMBER] = cmd[TASK_NUMBER] self.synchronization_cache = [] elif cmd[MSG] == DATA: data = cmd[DATA] self.synchronization_cache.extend(data) self.synchronization_message[TASK_HAD_DONE] += cmd[TASK_NUMBER] elif cmd[MSG] == "stop": if self.synchronization_message[TASK_HAD_DONE] == self.synchronization_message[TASK_NUMBER]: # 更新标准 train_info = cmd[DIFFICULTY_MES] self._set_train_info(train_info) # 删除所有人员信息 self.manager.delete_all_person() # 更新人员信息 format_container = [ { PERSON_FEATURE_LIST[key_index]: row[key_index] for key_index in range(0, len(PERSON_FEATURE_LIST)) } for row in self.synchronization_cache ] container_dir = {} for con in format_container: container_dir[con["id"]] = con format_container = container_dir.values() self.manager.insert_many_person_from_json(format_container) self.synchronization_message[STATUS] = STOP else: self.synchronization_cache = [] # 清空录制屏幕视频 video_recorder.delete_video() return jsonify({STATUS: STATUS_OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} print(traceback.format_exc()) return jsonify(error_msg) def score_synchronization_server(self): try: cmd = request.json if self.synchronization_message[STATUS] == STARTING: if cmd[MSG] == "start": self.synchronization_message[TASK_HAD_DONE] = 0 self.synchronization_message[TASK_NUMBER] = cmd[TASK_NUMBER] self.synchronization_cache = [] elif cmd[MSG] == DATA: data = cmd[DATA] self.synchronization_cache.extend(data) self.synchronization_message[TASK_HAD_DONE] += cmd[TASK_NUMBER] elif cmd[MSG] == "stop": if self.synchronization_message[TASK_HAD_DONE] == self.synchronization_message[TASK_NUMBER]: self.synchronization_cache.copy() format_container = [ { SCORE_FEATURE_LIST[key_index]: row[key_index] for key_index in range(0, len(SCORE_FEATURE_LIST)) } for row in self.synchronization_cache ] # container_dir = {} # for con in format_container: # container_dir[con["id"]] = con # format_container = container_dir.values() # 逐条成绩更新 for con in format_container: result = self.manager.select_a_score(person_id=con[ID], score_type=con['score_type'], batch=self.summary_info['batch']) if result is None: self.manager.insert_a_score(record=con["record"], score=con['score'], person_id=con[ID], score_type=con['score_type'], batch=self.summary_info['batch'], person_class=con['class'], name=con[NAME]) else: _record = float(result[0]['record']) _score = float(result[0]['score']) if con['score_type'] in {RUNAROUND, RUNNING} and con["record"] != -1 \ and (float(con["record"]) < _record or float(con['score']) > _score): self.manager.update_timekeeping_score(record=con["record"], score=con['score'], person_id=con[ID], score_type=con['score_type'], batch=self.summary_info['batch']) elif float(con["record"]) > _record or float(con['score']) > _score: self.manager.update_counting_score(record=con["record"], score=con['score'], person_id=con[ID], score_type=con['score_type'], batch=self.summary_info['batch']) self.synchronization_message[STATUS] = STOP else: self.synchronization_cache = [] return jsonify({STATUS: STATUS_OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} print(traceback.format_exc()) return jsonify(error_msg) def video_synchronization_server(self): try: cmd = request.json if self.synchronization_message[STATUS] == STARTING: if cmd[MSG] == OK: name = cmd["name"] v = base64.b64decode(cmd[DATA]) v_dir = os.path.join(GLOBAL_DIR, "LSZXVideo", "Video", name) with open(v_dir, 'wb') as f: f.write(v) self.synchronization_message[TASK_HAD_DONE] += 1 return jsonify({STATUS: STATUS_OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} print(traceback.format_exc()) return jsonify(error_msg) # 广播服务，用于转发设备间的信息 @staticmethod def broadcast(): try: json_mes = request.json targets = json_mes[TARGET] data = json_mes[DATA] # 创建一个最大并发量为16的线程池 executor = concurrent.futures.ThreadPoolExecutor(max_workers=16) futures = [] for url_target in targets: q = queue.Queue() # 创建队列，用于维护数据顺序 # 为每个URL创建的队列按顺序添加数据项 for seg_data in data: q.put(seg_data) # 将数据放入队列中 # 为每个URL创建一个任务 future = executor.submit(send_post_request, q, url_target) futures.append(future) concurrent.futures.wait(futures) # 等待所有任务完成 return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 开启成绩同步 def waiting_score_synchronization(self): self.synchronization_kill_sign = False self.reset_synchronization_status() # 检查网络链接 try: if not self.connection.get_connected_wifi() and not self.connection.master_mode: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络未链接！"} return jsonify(error_msg) except Exception as e: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络错误！", MSG: f"{e.args}"} return jsonify(error_msg) self_ip = self.connection.get_self_ip() if not self_ip: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "无法加入主机网络，请等待网络初始化，或重新链接WiFi！"} return jsonify(error_msg) try: self.synchronization_message[STATUS] = STARTING self.speak_driver.add_speak("准备开始同步数据！") while not self.synchronization_kill_sign: time.sleep(0.1) # 等待更新结束 if self.synchronization_message[STATUS] == STOP: break self.speak_driver.add_speak("同步完成！") return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 发送同步数据 def send_data_synchronization(self): self.synchronization_kill_sign = False self.reset_synchronization_status() # 检查网络链接 try: if not self.connection.get_connected_wifi() and not self.connection.master_mode: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络未链接！"} return jsonify(error_msg) except Exception as e: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络错误！", MSG: f"{e.args}"} return jsonify(error_msg) self.synchronization_message[STATUS] = STARTING # 获得所有设备ip ip_list = self.connection.get_other() try: # 获得所有人员信息 data = self.manager.select_all_person() format_data = [[row.get(key) for key in PERSON_FEATURE_LIST] for row in data] self.synchronization_message[TASK_NUMBER] = len(format_data) step = 100 counting = 0 # 发送起始包 start_pkg = {MSG: "start", TASK_NUMBER: len(format_data)} pkg = [start_pkg] # 发送所有数据 for segment_data in [format_data[i: i + step] for i in range(0, len(format_data), step)]: counting = len(segment_data) data_pkg = {MSG: DATA, DATA: segment_data, TASK_NUMBER: counting} pkg.append(data_pkg) self.synchronization_message[TASK_HAD_DONE] += len(segment_data) # 发送结束包 stop_pkg = {MSG: "stop", DIFFICULTY_MES: self.train_info} pkg.append(stop_pkg) targets = [ f"http://{ip}:{TERMINAL}/synchronization_server" for ip in ip_list ] master_ip = self.connection.get_master_ip() broadcast_pkg = { TARGET: targets, DATA: pkg } requests.post(url=f"http://{master_ip}:{TERMINAL}/broadcast", json=broadcast_pkg, timeout=30) # 更新同步状态包 self.synchronization_message[STATUS] = STOP self.speak_driver.add_speak("同步完成！") return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取数据同步进度 def get_synchronization_processing(self): mes_got = copy.deepcopy(self.synchronization_message) try: return jsonify({STATUS: OK, DATA: mes_got}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 结束数据同步 def stop_data_synchronization(self): try: self.synchronization_kill_sign = True self.reset_synchronization_status() return jsonify({STATUS: OK}) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 发送同步成绩 def send_score_synchronization(self): self.synchronization_kill_sign = False self.reset_synchronization_status() # 检查网络链接 try: if (not self.connection.get_connected_wifi() and not self.connection.master_mode): self.speak_driver.add_speak("网络未连接！") error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络未链接！"} return jsonify(error_msg) except Exception as e: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络错误！", MSG: f"{e.args}"} self.speak_driver.add_speak("网络错误！请重新链接WiFi！") return jsonify(error_msg) # 获得所有设备ip ip_list = self.connection.get_other() try: self.speak_driver.add_speak("准备开始同步数据！") # 更新同步状态包 self.synchronization_message[STATUS] = 1 self.synchronization_message[TASK_HAD_DONE] = 0 self.synchronization_message[TASK_NUMBER] = 0 # 获得所有成绩信息 data = self.manager.select_score() format_data = [[row.get(key) for key in SCORE_FEATURE_LIST] for row in data] video_num = len(os.listdir(video_path)) self.synchronization_message[TASK_NUMBER] = len(format_data) + video_num * 2 step = 100 counting = 0 # 发送起始包 start_pkg = {MSG: "start", TASK_NUMBER: len(format_data)} pkg = [start_pkg] for segment_data in [ format_data[i: i + step] for i in range(0, len(format_data), step) ]: counting += len(segment_data) data_pkg = {MSG: DATA, DATA: segment_data, TASK_NUMBER: counting} pkg.append(data_pkg) self.synchronization_message[TASK_HAD_DONE] += len(segment_data) # 压缩视频 try: if not os.path.exists(output_video_path): os.mkdir(output_video_path) else: shutil.rmtree(output_video_path) os.mkdir(output_video_path) input_video_list = os.listdir(video_path) for vide_path in input_video_list: if not self.synchronization_kill_sign: video_sending.convert_video(vide_path) self.synchronization_message[TASK_HAD_DONE] += 1 else: break targets = [ f"http://{ip}:{MANAGER}/video_synchronization_server" for ip in self.connection.get_manager() ] master_ip = self.connection.get_master_ip() for filename in os.listdir(output_video_path): if not self.synchronization_kill_sign: v_name = os.path.join(output_video_path, filename) with open(v_name, 'rb') as file: v = base64.b64encode(file.read()).decode('utf-8') file = [{"name": filename, DATA: v, MSG: OK}] broadcast_pkg = { TARGET: targets, DATA: file } requests.post(url=f"http://{master_ip}:{TERMINAL}/broadcast", json=broadcast_pkg, timeout=30) self.synchronization_message[TASK_HAD_DONE] += 1 else: break except: print(traceback.format_exc()) print("发送压缩视频失败！！！") # 发送结束包 stop_pkg = {MSG: "stop"} pkg.append(stop_pkg) targets = [ f"http://{ip}:{MANAGER}/score_synchronization_server" for ip in self.connection.get_manager() ] master_ip = self.connection.get_master_ip() broadcast_pkg = { TARGET: targets, DATA: pkg } requests.post(url=f"http://{master_ip}:{TERMINAL}/broadcast", json=broadcast_pkg, timeout=3 * len(ip_list) + 1) # 清空录制屏幕视频 # video_recorder.delete_video() # 更新同步状态包 self.synchronization_message[STATUS] = STOP self.speak_driver.add_speak("同步完成！") return jsonify({STATUS: OK}) except Exception as e: self.speak_driver.add_speak("同步时发生错误，同步结束！") print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 开启数据同步 def waiting_data_synchronization(self): self.synchronization_kill_sign = False self.reset_synchronization_status() # 检查网络链接 try: if not self.connection.get_connected_wifi() and not self.connection.master_mode: self.speak_driver.add_speak("网络未连接！") error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络未链接！"} return jsonify(error_msg) except Exception as e: error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "网络错误！", MSG: f"{e.args}"} self.speak_driver.add_speak("网络错误！请重新链接WiFi！") return jsonify(error_msg) self_ip = self.connection.get_self_ip() if not self_ip: self.speak_driver.add_speak("无法加入主机网络，请等待网络初始化，或重新链接WiFi！") error_msg = {STATUS: CONNECTION_ERROR, ERROR_MESSAGE: "无法加入主机网络，请等待网络初始化，或重新链接WiFi！"} return jsonify(error_msg) try: self.synchronization_message[STATUS] = STARTING self.speak_driver.add_speak("准备开始同步数据！") while not self.synchronization_kill_sign: time.sleep(0.1) # 等待更新结束 if self.synchronization_message[STATUS] == STOP: break self.speak_driver.add_speak("同步完成！") # 清空录制屏幕视频 video_recorder.delete_video() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取视频列表 def video_list(self): try: cmd = request.json person_id = cmd['id'] project = cmd['project'] response_data = [] for i in os.listdir(video_path): v_name = i.rsplit('.', 1)[0] data = v_name.split('_') if data[0] == person_id and data[2] == project: t = data[4].split('-') date = "{}年{}月{}日{}时{}分".format(t[0], t[1], t[2], t[3], t[4]) response_data.append({'name': i, 'date': date, 'project': data[2], 'score': data[3]}) response_data = sorted(response_data, key=lambda item: (item['date'], item['score']), reverse=True) return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取视频 def video_server(self): try: cmd = request.json file_name = cmd['file_name'] video_dir = os.path.join(video_path, file_name) return send_file(video_dir) except Exception as e: error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 向管理端发送成绩 def send_score(self): while True: try: self.send_score_signal.wait() self.send_score_signal.clear() manager_ip_list = self.connection.get_manager() ip = self.connection.get_self_ip() _ip = ip.split('.')[2] if manager_ip_list and _ip == '138': self.synchronization_info['time'] = time.time() broadcast_pkg = { DATA: self.synchronization_info, 'ip': ip } for manage_ip in manager_ip_list: try: requests.post(url=f"http://{manage_ip}:{MANAGER}/totals_synchronization_server", json=broadcast_pkg, timeout=3) except: continue except Exception: # print(traceback.format_exc()) continue # 向管理端发送数据库成绩 def send_sql_score(self): while True: try: self.send_sql_score_signal.wait() self.send_sql_score_signal.clear() manager_ip_list = self.connection.get_manager() ip = self.connection.get_self_ip() _ip = ip.split('.')[2] if manager_ip_list and _ip == "138": broadcast_pkg = { DATA: self._statistics_sql_score, 'ip': ip } for manager_ip in manager_ip_list: try: requests.post(url=f"http://{manager_ip}:{MANAGER}/totals_sql_synchronization_server", json=broadcast_pkg, timeout=3) except: continue self._statistics_sql_score = {} except Exception as e: # print(traceback.format_exc()) continue # 汇总成绩 def totals_synchronization_server(self): try: cmd = request.json ip = cmd['ip'] cmd['data']['time'] = time.time() if ip not in self._statistics_score: self._statistics_score.update({ip: {}}) self._statistics_score[ip] = cmd['data'] self.processing_score_signal.set() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 处理大屏实时成绩 def processing_score(self): while True: try: self.processing_score_signal.wait() self.processing_score_signal.clear() new_statistics_score = {SITUP: [], RUNAROUND: [], PUSHUP: [], PULLUP: [], RUNNING: []} for k, v in self._statistics_score.items(): for key, value in v.items(): if key == RUNNING: new_statistics_score.setdefault(key, []) new_statistics_score[key] = value new_statistics_score[key].sort(key=lambda x: (1 if x['normal'] else 0, x['rank'])) elif key != 'time': new_statistics_score.setdefault(key, []) if value: new_statistics_score[key].append(value) else: pass if key == 'runaround': new_statistics_score[key].sort(key=lambda x: (1 if x['normal'] else 0, x['count'])) else: new_statistics_score[key].sort(key=lambda x: (1 if x['normal'] else 0, -x['count'])) if key != 'time': for i in new_statistics_score[key]: if not i['normal'] and int(i['id']) not in self.closed_chamber: self.hrbo_info.update({int(i['id']): i}) self.statistics_score.update(new_statistics_score) # 5分钟后移出小黑屋 temp = self.closed_chamber.copy() for key, value in temp.items(): this_time = time.time() if this_time - int(value) > 300: del self.closed_chamber[key] except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 汇总数据库成绩 def totals_sql_synchronization_server(self): try: cmd = request.json exercise_tag = next(iter(cmd['data'])) if exercise_tag != RUNAROUND: record = str(exercise_tag) + '_' + 'count' score = str(exercise_tag) + '_' + 'score' else: record = 'run_bf_count' score = 'run_bf_score' for o in self.sql_data: for key, value in cmd['data'][exercise_tag].items(): if o['id'] == key: _record = self._sql_score[key][record] o[score] = value['score'] s = str(round(value['record'], 2)) if _record: o[record]['count'] = s + '(' + str(_record) + ')' else: o[record]['count'] = s o[record]['is_changed'] = True # if k in ('pullup_count', 'pullup_score', 'pushup_count', 'pushup_score', 'situp_count', # 'situp_score', 'running_score', 'run_bf_score'): # if n[k]: # if not o[k] or n[k] > o[k]: # o[k] = n[k] # elif k in ('run_bf_count', 'running_count'): # if n[k] and n[k] > 0: # if not o[k] or float(n[k]) < float(o[k]): # o[k] = format(float(n[k]), '.2f') # break threading.Thread(target=self.processing_sql).start() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 处理数据库成绩 def processing_sql(self): try: response_data = self.sql_data total = [ {'hr': row['hr'], 'bo': row['bo'], 'id': row['id'], 'name': row['name'], SITUP: row['situp_count'], RUNAROUND: row['run_bf_count'], PUSHUP: row['pushup_count'], PULLUP: row['pullup_count'], RUNNING: row['running_count']} for row in response_data] top_score = 0 top_situp = 0 top_pushup = 0 top_pullup = 0 top_runround = 40 top_running = 1800 for i in response_data: if i['pullup_score'] and i['situp_score'] and i['running_score'] and \ i['run_bf_score']: total_score = (float(i['pullup_score']) + float(i['situp_score']) + float(i['running_score']) + float(i['run_bf_score'])) / 4 if total_score > top_score: top_score = total_score self.statistics_score['top']['total'] = {'name': i['name'], 'score': top_score} if i['pullup_count']['count']: if '(' in str(i['pullup_count']['count']): count = float(i['pullup_count']['count'].split('(')[0]) if count > top_pullup: top_pullup = count self.statistics_score['top'][PULLUP] = {'name': i['name'], 'score': top_pullup} else: if float(i['pullup_count']['count']) > top_pullup: top_pullup = float(i['pullup_count']['count']) self.statistics_score['top'][PULLUP] = {'name': i['name'], 'score': top_pullup} if i['pushup_count']['count']: if '(' in str(i['pushup_count']['count']): count = float(i['pushup_count']['count'].split('(')[0]) if count > top_pushup: top_pushup = count self.statistics_score['top'][PUSHUP] = {'name': i['name'], 'score': top_pushup} else: if float(i['pushup_count']['count']) > top_pushup: top_pushup = float(i['pushup_count']['count']) self.statistics_score['top'][PUSHUP] = {'name': i['name'], 'score': top_pushup} if i['situp_count']['count']: if '(' in str(i['pushup_count']['count']): count = float(i['situp_count']['count'].split('(')[0]) if count > top_situp: top_situp = count self.statistics_score['top'][SITUP] = {'name': i['name'], 'score': top_situp} else: if float(i['situp_count']['count']) > top_situp: top_situp = float(i['situp_count']['count']) self.statistics_score['top'][SITUP] = {'name': i['name'], 'score': top_situp} if i['run_bf_count']['count']: if '(' in str(i['run_bf_count']['count']): count = float(i['run_bf_count']['count'].split('(')[0]) if top_runround > count > 0: top_runround = count self.statistics_score['top'][RUNAROUND] = {'name': i['name'], 'score': top_runround} else: if top_runround > float(i['run_bf_count']['count']) > 0: top_runround = float(i['run_bf_count']['count']) self.statistics_score['top'][RUNAROUND] = {'name': i['name'], 'score': top_runround} if i['running_count']['count']: if '(' in str(i['running_count']['count']): count = float(i['running_count']['count'].split('(')[0]) if count < top_running: top_running = count self.statistics_score['top'][RUNNING] = {'name': i['name'], 'score': top_running} else: if float(i['running_count']['count']) < top_running: top_running = float(i['running_count']['count']) self.statistics_score['top'][RUNNING] = {'name': i['name'], 'score': top_running} self.statistics_score['total'] = total except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) def statistics(self, ws): while True: try: time.sleep(0.5) self.statistics_score.update({STATUS: OK}) ws.send(json.dumps(self.statistics_score)) temp = self._statistics_score.copy() for k, v in temp.items(): this_time = time.time() if this_time - v['time'] > 60: del self._statistics_score[k] self.processing_score_signal.set() except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 接收心率血氧信息 def totals_HrBoInfo_server(self): try: cmd = request.json data = cmd["data"] # 5分钟后移出小黑屋 temp = self.closed_chamber.copy() for key, value in temp.items(): this_time = time.time() if this_time - int(value) > 300: del self.closed_chamber[key] # 汇总所有人hrbo信息 for k, v in data.items(): self._hrbo_info[k] = v # 筛选出警告人员 temp = self._hrbo_info.copy() for key, value in temp.items(): this_time = time.time() if this_time - value['record_time'] > 120: del self._hrbo_info[key] p_data = self.manager.select_a_person_from_band(key) if p_data: p_id = int(p_data[0]["id"]) p_name = p_data[0]["name"] for i in self.statistics_score['total']: if p_id == int(i[ID]): i['hr'] = value['hr'] i['bo'] = value['bo'] break if not value["normal"] and p_id not in self.closed_chamber: self.hrbo_info.update({p_id: value}) self.hrbo_info[p_id]["name"] = p_name return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 心率血氧有误 def abnormal(self, ws): report_tag = "" while True: try: time.sleep(0.5) if self.hrbo_info: data = [] for key, value in self.hrbo_info.items(): data.append({"name": value["name"], "id": key, HR: value[HR], BO: value[BO]}) # # 演示版 # if int(key) == 34: # data.append({"name": value["name"], "id": key, HR: value[HR]['hr'], BO: value[BO]['bo']}) data = sorted(data, key=lambda item: item['id']) new_report_tag = "@".join([str(item['id']) for item in data]) ws.send(json.dumps({STATUS: OK, "data": data})) if report_tag != new_report_tag: report_tag = new_report_tag beep(duration=1500, freq=640) self.speak_driver.add_speak("体征监测系统警报！") for d in data: self.speak_driver.add_speak(f"{d['name']}体征异常！请尽快救护！") else: report_tag = "" ws.send(json.dumps({STATUS: OK, "data": []})) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 心率血氧有误回传信号给后端 def receive(self): try: cmd = request.json self.hrbo_info = {} # 关入小黑屋 for i in cmd["data"]: this_time = time.time() p_id = int(i["id"]) self.closed_chamber.update({p_id: this_time}) return jsonify({STATUS: OK, "msg": "成功"}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 发送工程代码 def send_code(self): try: if FACTORY_MODE.is_set(): mes = "正在进行其他指令，请稍候！" self.speak_driver.add_speak(mes) return jsonify({STATUS: UNKNOWN_ERROR, "msg": mes}) else: FACTORY_MODE.set() cmd = request.json data = cmd["data"] code_re = re.search("[a-zA-Z0-9]+(?=@)", data) value_re = re.search("[0-9]+", data) code = code_re.group() if code_re else None value = value_re.group() if value_re else None mes = "指令不存在！" if code in FUNC_TABLE: mes = FUNC_TABLE[code](value=value, eb=self) else: self.speak_driver.add_speak(mes) return jsonify({STATUS: OK, "msg": mes}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 退出工程模式 def stop_code(self): try: FACTORY_MODE.clear() return jsonify({STATUS: OK, "msg": "成功"}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 演示视频1 def display_video1(self): try: video_dir = ".//display_video//1.mp4" return send_file(video_dir) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 演示视频2 def display_video2(self): try: video_dir = ".//display_video//2.mp4" return send_file(video_dir) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 演示视频3 def display_video3(self): try: video_dir = ".//display_video//3.mp4" return send_file(video_dir) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 演示视频4 def display_video4(self): try: video_dir = ".//display_video//4.mp4" return send_file(video_dir) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 读取当前考核批次GET def get_batch(self): try: batch = self.summary_info['batch'] return jsonify({STATUS: OK, DATA: batch}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 生成新的考核批次GET def set_batch(self): try: batch = self.manager.generate_batch() if batch: self.summary_info['batch'] = batch return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取所有的课目GET def get_all_projects(self): try: projects = [] for k, v in EXERCISE_NAME_LIST.items(): projects.append({'value': k, 'label': v}) return jsonify({STATUS: OK, 'projects': projects}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取所有的批次GET def get_all_batch(self): try: all_batch = self.manager.get_all_batch() return jsonify({STATUS: OK, 'batch': all_batch}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取所有的人员GET def get_all_people(self): try: all_name = [] for k, v in self.manager.get_all_name().items(): all_name.append({'id': k, 'name': v}) return jsonify({STATUS: OK, 'people': all_name}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获取所有的班级GET def get_all_classes(self): try: all_class = self.manager.get_all_class() return jsonify({STATUS: OK, 'class': all_class}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 汇总分析总览界面 # 获得课目训练汇总默认的课目和批次GET def get_projects_batch(self): try: projects = [] for k, v in self.summary_info['default_projects'].items(): projects.append(v) batch = self.summary_info['default_batch'] return jsonify({STATUS: OK, 'projects': projects, 'batch': batch}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置课目训练汇总课目和批次POST def set_projects_batch(self): try: cmd = request.json self.summary_info['default_projects'] = cmd['projects'] self.summary_info['default_batch'] = cmd['batch'] self._update_summary_info() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 环形图--总览 def circular_diagram(self): try: cmd = request.json projects = [] categorize = [] response_data = {} for k, v in cmd['projects'].items(): projects.append(k) response_data.update({k: []}) categorize.append({'name': v, 'keyname': k}) batch = cmd['batch'] data = self.manager.get_exercise_statistic( type_list=projects, batch=batch) for k, v in data.items(): for j in categorize: if k == j['keyname']: for i in GRADE_RANK: percentage = round(v[i]['percentage'] * 100, 2) response_data[k].append(percentage) return jsonify({STATUS: OK, 'categorize': categorize, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 详细成绩分布--总览环形图 def detailed_grade_info_1(self): try: cmd = request.json project = None for k, v in cmd['projects'].items(): project = k batch = cmd['batch'] grade = cmd['grade'] data = self.manager.get_exercise_statistic( type_list=[project], batch=batch ) response_data = [ { "id": person_data[ID], "name": person_data[NAME], "score": person_data[SCORE], "count": person_data[RECORD] } for person_data in data[project][grade][DATA] ] return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得各班课目训练汇总默认的课目和批次GET def get_class_projects_batch(self): try: projects = [] for k, v in self.summary_info['default_class_projects'].items(): projects.append(v) batch = self.summary_info['default_class_batch'] return jsonify({STATUS: OK, 'projects': projects, 'batch': batch}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置各班课目训练汇总课目和批次POST def set_class_projects_batch(self): try: cmd = request.json self.summary_info['default_class_projects'] = cmd['projects'] self.summary_info['default_class_batch'] = cmd['batch'] self._update_summary_info() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 班级条形图--总览 def class_bar_chart(self): try: cmd = request.json project = None for k, v in cmd['projects'].items(): project = k batch = cmd['batch'] data = self.manager.get_class_statistic( batch=batch, _type=project ) team = [] for k, v in data.items(): team.append(k) team = sorted(team) response_data = [[] for _ in GRADE_RANK] for index, item in enumerate(GRADE_RANK): for _index, _item in enumerate(team): percentage = round(data[_item][item]['percentage'] * 100, 2) response_data[index].append(percentage) return jsonify({STATUS: OK, 'class': team, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 详细成绩分布--总览班级条形图 def detailed_grade_info_2(self): try: cmd = request.json project = None for k, v in cmd['projects'].items(): project = k batch = cmd['batch'] grade = cmd['grade'] _class = cmd['class'] data = self.manager.get_class_statistic( batch=batch, _type=project ) response_data = [ { "id": person_data[ID], "name": person_data[NAME], "score": person_data[SCORE], "count": person_data[RECORD] } for person_data in data[_class][grade][DATA] ] return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 获得集体课目训练汇总默认的课目GET def get_group_projects(self): try: projects = [] for k, v in self.summary_info['default_group_projects'].items(): projects.append(v) return jsonify({STATUS: OK, 'projects': projects}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置集体课目训练汇总课目POST def set_group_projects(self): try: cmd = request.json self.summary_info['default_group_projects'] = cmd['projects'] self._update_summary_info() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 批次条形图--总览 def batch_bar_chart(self): try: cmd = request.json project = None for k, v in cmd['projects'].items(): project = k data = self.manager.get_group_statistic( _type=project ) all_batch = self.manager.get_all_batch() response_data = [[] for _ in GRADE_RANK] for index, item in enumerate(GRADE_RANK): for _index, _item in enumerate(all_batch): percentage = round(data[_item][item]['percentage'] * 100, 2) response_data[index].append(percentage) return jsonify({STATUS: OK, 'batch': all_batch, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 详细成绩分布--总览批次条形图 def detailed_grade_info_3(self): try: cmd = request.json project = None for k, v in cmd['projects'].items(): project = k batch = cmd['batch'] grade = cmd['grade'] data = self.manager.get_group_statistic( _type=project ) response_data = [ { "id": person_data[ID], "name": person_data[NAME], "score": person_data[SCORE], "count": person_data[RECORD] } for person_data in data[batch][grade][DATA] ] return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 汇总分析集体界面 # 获得集体汇总默认的课目、班级、批次 def get_team_projects_batch_class(self): try: projects = None for k, v in self.summary_info['default_team_project'].items(): projects = v batch = self.summary_info['default_team_batch'] team = self.summary_info['default_team_class'] return jsonify({STATUS: OK, 'projects': projects, 'batch': batch, 'class': team}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置集体汇总默认的课目、班级、批次 def set_team_projects_batch_class(self): try: cmd = request.json self.summary_info['default_team_project'] = cmd['projects'] self.summary_info['default_team_batch'] = cmd['batch'] self.summary_info['default_team_class'] = cmd['class'] self._update_summary_info() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 集体比例分布图 def scale_drawing(self): try: cmd = request.json project = None for k, v in cmd['projects'].items(): project = k team = cmd['class'] batch = cmd['batch'] group_data = self.manager.get_group_multiple_statistic( _type=project, class_list=team, batch=batch ) response_data = [ { "name": grade, "value": len(group_data[grade][DATA]), "children": [ { "id": data[ID], "name": data[NAME], "score": data[SCORE], "count": data[RECORD] if data[RECORD] else "无", "cn_name": GRADE_LIST[grade], "value": 1 } for data in group_data[grade][DATA] ] } for grade in GRADE_LIST ] return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) ## 汇总分析个人界面 # 获得个人汇总默认的课目、人员、批次 def get_personal_projects_batch_person(self): try: projects = [] for k, v in self.summary_info['default_personal_projects'].items(): projects.append(v) batch = self.summary_info['default_personal_batch'] personal = self.summary_info['default_personal_personal'] return jsonify({STATUS: OK, 'projects': projects, 'batch': batch, 'personal': personal}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 设置个人汇总默认的课目、人员、批次 def set_personal_projects_batch_person(self): try: cmd = request.json self.summary_info['default_personal_projects'] = cmd['projects'] self.summary_info['default_personal_batch'] = cmd['batch'] self.summary_info['default_personal_personal'] = cmd['person'] self._update_summary_info() return jsonify({STATUS: OK}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 雷达图 def radar_chart(self): try: cmd = request.json person = cmd['person'] projects = [] for k, v in cmd['projects'].items(): projects.append(k) batch = cmd['batch'] data = self.manager.get_individual_statistic( _id=str(person['id']), _type_list=projects, batch_list=batch ) response_data = data['type_avg'] return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg) # 折线图 def line_chart(self): try: cmd = request.json person = cmd['person'] projects = [] categorize = [] valueDdata = {} for k, v in cmd['projects'].items(): projects.append(k) valueDdata.update({k: []}) categorize.append({'name': v, 'keyname': k}) batch = cmd['batch'] data = self.manager.get_individual_statistic( _id=str(person['id']), _type_list=projects, batch_list=batch ) batch_data = data['batch'] batch_avg = data['batch_avg'] maxScode = 0 minScore = 0 average = [] for i in batch_data.values(): for k, v in valueDdata.items(): valueDdata[k].append(i[k]['score']) if i[k]['score'] > maxScode: maxScode = i[k]['score'] remainder = maxScode % 10 maxScode = maxScode + (10 - remainder) if maxScode < 100: maxScode = 100 interval = maxScode / 5 for k, v in batch_avg.items(): average.append(v) response_data = { 'minScore': minScore, 'maxScore': maxScode, 'interval': interval, 'average': average, 'barData': { 'categorize': categorize, 'valueDdata': valueDdata } } return jsonify({STATUS: OK, DATA: response_data}) except Exception as e: print(traceback.format_exc()) error_msg = {STATUS: UNKNOWN_ERROR, ERROR_MESSAGE: "出现错误，请联系后端开发人员！错误原因：{}".format(e)} return jsonify(error_msg)