import json import os from flask import Flask, request from flask_cors import * from PIL import Image import base64 from analyse_func import do_analysis from classify import api_dyna from db_project import get_status, process_db, lookup_data, add_new_data_dyna, add_new_data, create_table_dynamic, \ create_table, create_db import shutil from demo_func import run_3d_for_video # str_res="" # a dict contains the txt dict_total = {} li_last = [] # store dic_data's json format bJson = "" global_index = [6] global_video_index = 0 li_return = [] img_return = [] flag_run = False front_end_dir = "/root/www/dist" ##jump ##result["jump_power"]["level"]/["path"]/["val"] ##result["core_strength"]["level"]/["path"] 1 2 3 4 ##result["landing_position"]["level"]/["path"]["val"] ##juanfu ## result["head_power"]["level"]/["path"]/["val"] # result["core_strength"]["level"]/["path"] ##result["body_stability"]["level"]/["val"] ##pingban """ result["arm_strength"]["level"]/["path"]/["val"] result["stability"]["level"]/["path"]/["val"] result["standard"]["level"]/["path"]/["val"] """ ##shendun """ result["leg_strength"]["level"]/["path"] """ ##gaotaitui """ result["standard"]["level"]/["path"] """ # read txt file and change it to the list format def read_pose_processes_txt_result(): file_dir = "pose_processed_txt" # the file's path dir_list = os.listdir(file_dir) # dir_list=["1.txt","2.txt","3.txt","4.txt"] global li_last # 分析的最后结果 global dict_total # list的字典 li_last = [] dict_total = {} for cur_file in dir_list: if (cur_file[0:len(cur_file) - 4] != str(global_index[0])): # print(cur_file[0:len(cur_file)-4]) # print(global_index[0]) continue path = os.path.join(file_dir, cur_file) # path="pose_processed_txt/1.txt" f = open(path, encoding='utf-8') lines = f.readlines() # process logic,{"正面分析"：{“头部分析”:{}，"肩部分析": {}} dic_data = {} for i in range(len(lines)): if lines[i][0] == '-': continue elif lines[i][0] == "正": dic_data = {"title": "正面分析如下"} li = [] dic_1 = {} i = i + 1 # i是头部分析那行 dic_1["name"] = "头部分析" dic_1["value"] = lines[i][5:len(lines[i]) - 1] i = i + 1 # 头部数字级别 dic_1["level"] = int(lines[i].strip()) li.append(dic_1) dic_2 = {} i += 2 # 第六行 i=6 dic_2["name"] = "肩部分析" # num_risk_1 = int(lines[i + 1].strip()) dic_2["value"] = lines[i][:-1] + " " + lines[(i + 1)][:-1] num_risk_2 = int(lines[i + 2].strip()) dic_2["level"] = int(num_risk_2) li.append(dic_2) i += 4 # i=10 dic_3 = {} dic_3["name"] = "髋部分析" dic_3["value"] = lines[i][:-1] + " " + lines[(i + 1)][:-1] num_risk_1 = int(lines[i + 2].strip()) # num_risk_2 = int(lines[i + 3].strip()) dic_3["level"] = int(max(num_risk_1, 0)) i += 4 # i=14 dic_4 = {"name": "腿型分析", "value": lines[i][:-1]} li.append(dic_3) li.append(dic_4) dic_data["detail"] = li if ("非" in lines[i][:-1]): dic_4["level"] = 0 else: dic_4["level"] = 1 li_last.append(dic_data) elif lines[i][0] == "侧": li = [] dic_data = {"title": "侧面分析如下"} dic_1 = {} i += 1 dic_1["name"] = "颈椎分析" if lines[i][4] != ":": dic_1["value"] = lines[i][0:len(lines[i]) - 1] else: dic_1["value"] = lines[i][5:len(lines[i]) - 1] dic_1["level"] = int(lines[i + 1].strip()) i += 2 dic_2 = {"name": "站姿分析", "value": lines[i][5:len(lines[i]) - 1]} dic_2["level"] = int(lines[i + 1].strip()) i += 2 dic_3 = {"name": "髋部分析", "value": lines[i][5:len(lines[i]) - 1]} dic_3["level"] = int(lines[i + 1].strip()) i += 2 dic_4 = {"name": "膝盖分析", "value": lines[i][7:len(lines[i]) - 1]} dic_4["level"] = int(lines[i + 1].strip()) li.append(dic_1) li.append(dic_2) li.append(dic_3) li.append(dic_4) dic_data["detail"] = li li_last.append(dic_data) else: i += 1 print(li_last) dict_total[cur_file] = li_last f.close() global li_return li_return = dict_total[(str(global_index[0]) + ".txt")] def run_the_process(index: int = 0, height: float = 110): # sys.path.append("../analyse_func.py") # import analyse_func # print(global_index[0]) do_analysis(global_index[0], height) # 把list转化为可以直接返回的json格式 def list_to_json(li): # print(li) aJson = json.dumps(li, ensure_ascii=False) print(aJson) # dict1={} # dict1["front"]="/home/seecoder/Documents/pose-correction/pc-demo/pose_processed_images/"+str(global_index[0])+"-0-result.jpg" # dict1["right"] = "/home/seecoder/Documents/pose-correction/pc-demo/pose_processed_images/" + \ # str(global_index[0] )+ "-1-result.jpg" # print(dict1) return aJson # print(dic_data) app = Flask(__name__) def save_file(filed): base = "pose_source_images/" filename = filed.filename index = global_index[0] # global_index[0] = index suffix = ".png" if filename.find("jpg") > 0: suffix = ".jpg" elif filename.find("png") > 0: suffix = ".png" save_name = "" print(str(index) + "index") print(filename + "filename") if "front" in filename: save_name = base + str(index) + str("-0") + suffix print(filename + "filename_front") print(save_name + "savename_front") elif "right" in filename: save_name = base + str(index) + str("-1") + suffix print(filename + "filename_right") print(save_name + "savename_right") else: print("nope") filed.save(save_name) print(filename) print(save_name) # 运行动态分析的入口 def run_video(mode, num, ana=1): q = "./video/" + str(num) + ".mp4" re = api_dyna(q, mode, ana) process_db(num) return re def lookup_video(num, ana=0): q = "./video/" + str(num) + ".mp4" mode = lookup_data(num) re = api_dyna(q, mode, ana) process_db(num) return re ###send a video and return a id marked the video (identical) @app.route('/get_videoid', methods=['POST']) @cross_origin() def get_videoid(): dict_video_ = request.form.to_dict() dict_video_mp4 = request.files.to_dict() ff = dict_video_mp4["video"] mode = dict_video_["mode"] global global_video_index global_video_index = int(add_new_data_dyna(mode)) base = "./video/" filename = base + str(global_video_index) + ".mp4" ff.save(filename) return {"id": str(global_video_index)} @app.route('/get_video_data', methods=['POST']) @cross_origin() def get_video_data(): dict_video_ = request.form.to_dict() global global_video_index global_video_index = int(dict_video_["id"]) # print(global_video_index) mode = str(lookup_data(global_video_index)) # get database mode # print(mode) re = run_video(mode, global_video_index, 1) return re @app.route('/get_video_status', methods=['POST']) @cross_origin() def get_video_status(): dict_video_ = request.form.to_dict() global global_video_index global_video_index = int(dict_video_["id"]) # bool = get_status(global_video_index) == 1 return {"state": get_status(global_video_index) == 1} @app.route('/query_video', methods=['POST']) @cross_origin() def query_video(): dict_video_ = request.form.to_dict() print(dict_video_) global global_video_index global_video_index = int(dict_video_["id"]) if get_status(global_video_index) == 0: mode = str(lookup_data(global_video_index)) # get database mode run_video(mode, global_video_index, 1) return lookup_video(global_video_index, 0) def get_3D(id): video_path = "video/" + id + ".mp4" run_3d_for_video(video_path) @app.route('/query_3D', methods=['POST']) @cross_origin() def query_3D(): id = request.form.to_dict()["id"] video_path = "video/" + id + "_reconstructed.mp4" saved_video_path = os.path.join(front_end_dir, os.path.basename(video_path)) if not os.path.exists(saved_video_path): get_3D(id) shutil.copyfile(video_path, saved_video_path) print(saved_video_path) return { 'url': saved_video_path.replace(front_end_dir, "http://106.15.1.178:8001"), # 'status':'finished' } # result={xx:xx,xx:xx} # result={power:{val:xx,path:xx},strength:{val:xx,path:xx}} # 1.锻炼建议 # 2.发送视频 # 3. @app.route('/send_form', methods=['POST']) @cross_origin() def send_form(): text_data = request.form.to_dict() fileList = request.files.to_dict() global global_index global_index[0] = int(add_new_data()) for filed in fileList.values(): save_file(filed) height = float(text_data["height"]) if text_data["height"] != "undefined" else None run_the_process(global_index[0], height) process_db(int(global_index[0])) return { "id": global_index[0], "status": True if height is not None else False } def get_base64(path): with open(path, "rb") as f: base64_data = base64.b64encode(f.read()) return base64_data.decode() @app.route('/get_result', methods=['POST']) @cross_origin() def run_project(): id = request.form.to_dict()["id"] print(id) base = "pose_processed_images" global_index[0] = int(id) read_pose_processes_txt_result() print(base + "/" + str(global_index[0]) + "-0-result.png") return { "front": get_base64(base + "/" + str(global_index[0]) + "-0-result.png"), "right": get_base64(base + "/" + str(global_index[0]) + "-1-result.png"), "data": list_to_json(li_return), } if __name__ == '__main__': # read_pose_processes_txt_result() # print(list_to_json(li_return)) # app.run() # app.run(host="0.0.0.0", port=8090, debug=True) app.run(host="0.0.0.0", port=8090) CORS(app, resouces=r'/*') # run_video("gaotaitui",19,1) # lookup_video(41, 0)