Spaces:
Paused
Paused
| import cv2, math | |
| import json, os, torch | |
| import numpy as np | |
| from sklearn.preprocessing import Normalizer | |
| from align import align_filter | |
| def merge_intervals_with_breaks(time_intervals, errors, max_break=1.5): | |
| print(f"时间区间: {time_intervals}") | |
| print(f"错误: {errors}") | |
| if not time_intervals: | |
| return [] | |
| # Sort intervals based on starting times (not necessary here as input is sorted but good practice) | |
| sorted_intervals = sorted(zip(time_intervals, errors), key=lambda x: x[0][0]) | |
| merged_intervals = [] | |
| current_interval, current_error = sorted_intervals[0] | |
| for (start, end), error in sorted_intervals[1:]: | |
| # Check if the current interval error is the same and the break between intervals is <= 1.5 seconds | |
| if error == current_error and start - current_interval[1] <= max_break: | |
| # Merge intervals | |
| current_interval = (round(current_interval[0]), round(max(current_interval[1], end))) | |
| else: | |
| # Save the completed interval | |
| merged_intervals.append(((round(current_interval[0]), round(current_interval[1])), current_error)) | |
| # merged_intervals.append((current_interval, current_error)) | |
| # Start a new interval | |
| current_interval, current_error = (round(start), round(end)), error | |
| # Add the last interval | |
| merged_intervals.append((current_interval, current_error)) | |
| return merged_intervals | |
| def findcos_single(k1, k2): | |
| u1 = np.array(k1).reshape(-1, 1) | |
| u2 = np.array(k2).reshape(-1, 1) | |
| source_representation, test_representation = u1, u2 | |
| a = np.matmul(np.transpose(source_representation), test_representation) | |
| b = np.sum(np.multiply(source_representation, source_representation)) | |
| c = np.sum(np.multiply(test_representation, test_representation)) | |
| # return 1 - (a / (np.sqrt(b) * np.sqrt(c))) | |
| cosine_similarity = a / (np.sqrt(b) * np.sqrt(c)) | |
| return 100 * (1 - (1 - cosine_similarity) / 2), 0 | |
| def align_hstack(frame_00, frame_01, keypoints_01=None): | |
| height_00 = frame_00.shape[0] | |
| height_01 = frame_01.shape[0] | |
| if height_01 != height_00: | |
| # 计算缩放比例，确保高度与 frame_00 一致 | |
| scale_factor = height_00 / height_01 | |
| new_width = int(frame_01.shape[1] * scale_factor) | |
| # 使用 OpenCV 的 resize 函数按比例缩放 frame_01 | |
| frame_01_resized = cv2.resize(frame_01, (new_width, height_00)) | |
| else: | |
| frame_01_resized = frame_01 | |
| # 现在可以水平拼接两个数组 | |
| combined_frame_ori = np.hstack((frame_00, frame_01_resized)) | |
| if keypoints_01 == None: return combined_frame_ori, None | |
| scale_factor = frame_00.shape[0] / frame_01.shape[0] # 根据高度的缩放比例 | |
| # 对 frame_01 的关键点进行缩放 | |
| keypoints_01_scaled = [] | |
| for point in keypoints_01: | |
| scaled_point = [point[0] * scale_factor, point[1] * scale_factor] # 仅对 x 和 y 坐标进行缩放 | |
| keypoints_01_scaled.append(scaled_point) | |
| return combined_frame_ori, keypoints_01_scaled | |
| def findCosineSimilarity_1(keypoints1, keypoints2): | |
| # transformer = Normalizer().fit(keypoints1) | |
| # keypoints1 = transformer.transform(keypoints1) | |
| user1 = np.concatenate((keypoints1[5:13], keypoints1[91:133]), axis=0).reshape(-1, 1) | |
| # transformer = Normalizer().fit(keypoints2) | |
| # keypoints2 = transformer.transform(keypoints2) | |
| user2 = np.concatenate((keypoints2[5:13], keypoints2[91:133]), axis=0).reshape(-1, 1) | |
| ####ZIYU | |
| source_representation, test_representation = user1, user2 | |
| a = np.matmul(np.transpose(source_representation), test_representation) | |
| b = np.sum(np.multiply(source_representation, source_representation)) | |
| c = np.sum(np.multiply(test_representation, test_representation)) | |
| # return 1 - (a / (np.sqrt(b) * np.sqrt(c))) | |
| cosine_similarity = a / (np.sqrt(b) * np.sqrt(c)) | |
| return 100 * (1 - (1 - cosine_similarity) / 2), 0 | |
| def load_json(path): | |
| with open(path, 'r') as file: | |
| return json.load(file) | |
| def eval(test, standard, tmpdir): | |
| test_p = tmpdir + "/user.mp4" | |
| standard_p = tmpdir + "/standard.mp4" | |
| os.system('python inferencer_demo.py ' + test_p + ' --pred-out-dir ' + tmpdir) # produce user.json | |
| scores = [] | |
| align_filter(tmpdir + '/standard', tmpdir + '/user', tmpdir) # 帧对齐 produce aligned vedios | |
| data_00 = load_json(tmpdir + '/standard.json') | |
| data_01 = load_json(tmpdir + '/user.json') | |
| cap_00 = cv2.VideoCapture(standard_p) | |
| cap_01 = cv2.VideoCapture(test_p) | |
| # Define keypoint connections for both videos (example indices, you'll need to customize) | |
| connections1 = [(9,11), (7,9), (6,7), (6,8), (8,10), (7,13), (6,12), (12,13)] | |
| connections2 = [(130,133), (126,129), (122,125), (118,121), (114,117), (93,96), (97,100), (101,104), (105,108), (109,112)] | |
| # Determine the minimum length of JSON data to use | |
| min_length = min(len(data_00), len(data_01)) | |
| frame_width = int(cap_00.get(cv2.CAP_PROP_FRAME_WIDTH)) | |
| frame_height = int(cap_00.get(cv2.CAP_PROP_FRAME_HEIGHT)) | |
| out = cv2.VideoWriter(tmpdir + '/output.mp4', cv2.VideoWriter_fourcc(*'XVID'), 5, (frame_width*2, frame_height*2)) | |
| cap_00.set(cv2.CAP_PROP_POS_FRAMES, 0) # 初始化视频从头开始读取 | |
| cap_01.set(cv2.CAP_PROP_POS_FRAMES, 0) | |
| comments = -1 | |
| error_dict = {} | |
| cnt = 0 | |
| line_width = 1 if frame_width // 300 == 0 else frame_width // 300 | |
| # 开始逐帧处理两个视频 | |
| while True: | |
| ret_00, frame_00 = cap_00.read() # 逐帧读取标准视频和用户视频的当前帧 | |
| ret_01, frame_01 = cap_01.read() | |
| if not ret_00 and ret_01: | |
| comments = 0 #.append("请尝试加快手势的完成速度，并确保每个动作都清晰可见。") | |
| break # Stop if either video runs out of frames | |
| elif ret_00 and not ret_01: | |
| comments = 1 #.append("请尝试放慢手势的完成速度，确保每个动作都清晰可见。") | |
| break # Stop if either video runs out of frames | |
| elif not ret_00 and not ret_01: | |
| comments = 2 | |
| break | |
| # combined_frame_ori = np.hstack((frame_00, frame_01)) | |
| # 获取视频当前的帧号 | |
| frame_id_00 = int(cap_00.get(cv2.CAP_PROP_POS_FRAMES)) | |
| frame_id_01 = int(cap_01.get(cv2.CAP_PROP_POS_FRAMES)) | |
| if frame_id_01 < min_length: | |
| combined_frame_ori, keypoints_01_scaled = align_hstack(frame_00, frame_01, data_01[frame_id_01]["instances"][0]["keypoints"]) | |
| else: | |
| combined_frame_ori, _ = align_hstack(frame_00, frame_01) | |
| # 处理标准视频中的关键点，并绘制关键点连接 | |
| if frame_id_00 < min_length: | |
| keypoints_00 = data_00[frame_id_00]["instances"][0]["keypoints"] | |
| for (start, end) in connections1: | |
| start = start - 1 | |
| end = end - 1 | |
| if start < len(keypoints_00) and end < len(keypoints_00): | |
| start_point = (int(keypoints_00[start][0]), int(keypoints_00[start][1])) | |
| end_point = (int(keypoints_00[end][0]), int(keypoints_00[end][1])) | |
| cv2.line(frame_00, start_point, end_point, (255, 0, 0), line_width) # (BGR) Blue line | |
| for (start, end) in connections2: | |
| start = start - 1 | |
| end = end - 1 | |
| for i in range(start, end): | |
| if i < len(keypoints_00) and i + 1 < len(keypoints_00): | |
| start_point = (int(keypoints_00[i][0]), int(keypoints_00[i][1])) | |
| end_point = (int(keypoints_00[i + 1][0]), int(keypoints_00[i + 1][1])) | |
| cv2.line(frame_00, start_point, end_point, (255, 0, 0), line_width) # Blue line | |
| # 将部分关键点保存并绘制圆点，标记关键位置 | |
| keypoints_00_ori = keypoints_00 | |
| keypoints_00 = keypoints_00[5:13] + keypoints_00[91:133] | |
| for point in keypoints_00: | |
| cv2.circle(frame_00, (int(point[0]), int(point[1])), 1, (0, 210, 0), -1) | |
| # 处理用户视频中的关键点，并进行误差分析 | |
| if frame_id_01 < min_length: | |
| error = [] | |
| bigerror = [] | |
| keypoints_01 = keypoints_01_scaled #data_01[frame_id_01]["instances"][0]["keypoints"] | |
| for (start, end) in connections1: | |
| start = start - 1 | |
| end = end - 1 | |
| if start < len(keypoints_01) and end < len(keypoints_01): | |
| start_point = (int(keypoints_01[start][0]), int(keypoints_01[start][1])) | |
| end_point = (int(keypoints_01[end][0]), int(keypoints_01[end][1])) | |
| cur_score = findcos_single([[int(keypoints_01[start][0]), int(keypoints_01[start][1])], [int(keypoints_01[end][0]), int(keypoints_01[end][1])]], [[int(keypoints_00_ori[start][0]), int(keypoints_00_ori[start][1])], [int(keypoints_00_ori[end][0]), int(keypoints_00_ori[end][1])]]) | |
| # 如果当前相似度小于 99.3，认为有误差，并记录下来 | |
| if float(cur_score[0]) < 98.8 and start != 5: | |
| error.append(start) | |
| cv2.line(frame_01, start_point, end_point, (0, 0, 255), 2) # Red line | |
| # 如果相似度低于 98，记录为大误差 | |
| if float(cur_score[0]) < 97.8: | |
| bigerror.append(start) | |
| else: | |
| cv2.line(frame_01, start_point, end_point, (255, 0, 0), line_width) # Blue line | |
| for (start, end) in connections2: | |
| start = start - 1 | |
| end = end - 1 | |
| for i in range(start, end): | |
| if i < len(keypoints_01) and i + 1 < len(keypoints_01): | |
| start_point = (int(keypoints_01[i][0]), int(keypoints_01[i][1])) | |
| end_point = (int(keypoints_01[i + 1][0]), int(keypoints_01[i + 1][1])) | |
| cur_score = findcos_single([[int(keypoints_01[i][0]), int(keypoints_01[i][1])], [int(keypoints_01[i + 1][0]), int(keypoints_01[i + 1][1])]], [[int(keypoints_00_ori[i][0]), int(keypoints_00_ori[i][1])], [int(keypoints_00_ori[i + 1][0]), int(keypoints_00_ori[i + 1][1])]]) | |
| if float(cur_score[0]) < 98.8: | |
| error.append(start) | |
| cv2.line(frame_01, start_point, end_point, (0, 0, 255), 2) # Red line | |
| if float(cur_score[0]) < 97.8: | |
| bigerror.append(start) | |
| else: | |
| cv2.line(frame_01, start_point, end_point, (255, 0, 0), line_width) # Blue line | |
| # 将用户视频的关键点绘制为圆点 | |
| keypoints_01 = keypoints_01[5:13] + keypoints_01[91:133] | |
| for point in keypoints_01: | |
| cv2.circle(frame_01, (int(point[0]), int(point[1])), 1, (0, 210, 0), -1) | |
| # Concatenate the images horizontally to display side by side | |
| # combined_frame = np.hstack((frame_00, frame_01)) | |
| combined_frame, _ = align_hstack(frame_00, frame_01) | |
| if frame_id_00 < min_length and frame_id_01 < min_length: | |
| min_cos, min_idx = findCosineSimilarity_1(data_00[frame_id_00]["instances"][0]["keypoints"], data_01[frame_id_01]["instances"][0]["keypoints"]) | |
| # 如果存在误差，将误差部分对应的人体部位加入内容列表 | |
| if error != []: | |
| # print(error) | |
| content = [] | |
| for i in error: | |
| if i in [5,7]: content.append('Left Arm') | |
| if i in [6,8]: content.append('Right Arm') | |
| if i > 90 and i < 112: content.append('Left Hand') | |
| if i >= 112: content.append('Right Hand') | |
| part = "" | |
| # 在视频帧上显示检测到的误差部位 | |
| # cv2.putText(combined_frame, "Please check: ", (int(frame_width*1.75), int(frame_height*0.2)), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 2) | |
| start_x = int(frame_width*1.75) + 10 #435 # 起始的 x 坐标 | |
| start_y = int(frame_height*0.2) + 50 # 45 | |
| line_height = 50 # 每一行文字的高度 | |
| # 将每一个部位的内容绘制到帧上 | |
| for i, item in enumerate(list(set(content))): | |
| text = "- " + item | |
| y_position = start_y + i * line_height | |
| # cv2.putText(combined_frame, text, (start_x, y_position), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 2) | |
| # big | |
| if bigerror != []: | |
| bigcontent = [] | |
| for i in bigerror: | |
| if i in [5,7]: bigcontent.append('Left Arm') | |
| if i in [6,8]: bigcontent.append('Right Arm') | |
| if i > 90 and i < 112: bigcontent.append('Left Hand') | |
| if i >= 112: bigcontent.append('Right Hand') | |
| # 记录当前帧的严重误差部位，存入 error_dict 中 | |
| error_dict[cnt] = list(set(bigcontent)) | |
| cnt += 1 | |
| combined_frame = np.vstack((combined_frame_ori, combined_frame)) | |
| out.write(combined_frame) | |
| scores.append(float(min_cos)) # 记录每一帧的相似度得分 | |
| fps = 5 # Frames per second | |
| frame_numbers = list(error_dict.keys()) # List of frame numbers 获取含有严重误差的帧号列表 | |
| time_intervals = [(frame / fps, (frame + 1) / fps) for frame in frame_numbers] # 将帧号转换为时间区间（秒） | |
| errors = [error_dict[frame] for frame in frame_numbers] # 每一帧对应的严重误差部位 | |
| final_merged_intervals = merge_intervals_with_breaks(time_intervals, errors) # 合并相邻或相近的时间区间，并记录对应的误差部位 | |
| out.release() | |
| # 返回三个结果： | |
| # 1. scores 的平均值，作为整体手势相似度的评分 | |
| # 2. final_merged_intervals，合并后的误差时间区间及其对应的误差信息 | |
| # 3. comments，用于给用户的速度建议（加快或放慢手势） | |
| return sum(scores) / len(scores), final_merged_intervals, comments | |
| def install(): | |
| import subprocess | |
| subprocess.run(["pip", "uninstall", "-y", "numpy"], check=True) | |
| subprocess.run(["pip", "install", "numpy<2"], check=True) | |
| os.system('mim install mmengine') | |
| os.system('mim install mmcv==2.2.0') | |
| os.system('git clone https://github.com/open-mmlab/mmpose.git') | |
| os.chdir('mmpose') | |
| os.system('pip install -r requirements.txt') | |
| os.system('pip install -v -e .') | |
| os.chdir('../') | |
| os.system('git clone https://github.com/open-mmlab/mmdetection.git') | |
| os.system('cp ./test.py ./mmdetection/mmdet/__init__.py') | |
| os.chdir('mmdetection') | |
| os.system('pip install -v -e .') | |
| os.chdir('../') | |
| os.system('apt-get install ffmpeg imagemagick') | |