迳口麒麟舞(KirinDance)AI教练人体姿态动作动态评分算法

迳口麒麟舞(KirinDance)AI教练人体姿态动作动态评分算法

文章目录

关注作者：JuruoAssociation 原创于 CSDN
原创不易，谢绝转载。点击支持原文

为了开发麒麟舞AI动作教练，动作解算我们采用的是OpenPose和PoseNet，但由于网上没有较为完善的动作评分系统，我们决定自行开发。
首先，我们的接口可以从摄像头中得到各个关节的置信度（Confidence）和位置（X、Y），设置一定的阈值（threshold）具有标记阈值之上置信度的关节为出现的关节，评分则基于这些出现的关节进行评分。
原项目已经开源在github

第一版

算法原理

第一版算法我们主要研究关节的瞬态位置，对于一个动作，选取若干个关节，获取其位置信息，然后做傅里叶变换转换到频域，这样即使是时间不同步（考虑显示延时和用户反应时间），他们的幅度曲线也具有相似性。

我们需要解算的是图1.A的标准视频中的静态动作数据，以及用户的摄像头动作流数据，图中横轴为时间，纵轴为位置（X坐标或Y坐标）。如果对于图一A和图一B，最终得到的频域曲线是相似的，随机延迟会被消除。评分依据是正确频率上的幅度和错误频率上的幅度。对于图一C，最终的频域曲线幅度将变小，对于图一D，频率将不正确。

局限性

但局限性也十分明显，如上的算法需要测量的频率极低，如果样本步长过小，①会导致精度丢失，②小幅度波动测量的十分清楚，导致产生大量杂值，然而由于动作解算方法的局限性，对于一个2160p的摄像头，10px以内的波动都是正常情况。③以上的方法对用户不友好，用户没办法获得自己的动作反馈，而且评分反馈难以让用户理解。④由于FFT对步长的要求，这是算法系统延时限制，会导致评分不能及时展示，最好的方案是等到全部录制完毕再评分，可能影响用户积极性。

解决方案

对于问题②可以采用低通滤波的方式解决，但是①③④暂时没有更好的方案。因此我们在第一版基础上没有做过多停留，改善到第二版进行实现。

第二版

算法原理

后来我们经过讨论，使用了一种模拟算法，这种模拟算法具体计算步骤如下：

1. 将图像从BGR格式转换为RGB格式。 使用 MediaPipe 的 pose 模型处理图像，获取姿态 landmarks。
2. 如果未检测到姿态 landmarks，返回 None。 定义 get_coords 函数，用于从 landmarks 中提取坐标。
3. 提取左右肩膀、手肘、手腕、臀部、膝盖和脚踝的坐标。 计算中心肩膀和中心臀部的坐标，以及垂直参考点的坐标。
4. 计算左右手臂、左右腿和身体倾斜的角度。 返回计算得到的角度列表。 如果过程中发生异常，打印错误信息并返回 None。

以上4个步骤可以得到图像中人体关键点的角度，包括左右手臂、左右腿和身体倾斜的角度，具体代码如下：

# Function to calculate angledefcalculate_angle(fp, sp, tp):
    fp, sp, tp = np.array(fp), np.array(sp), np.array(tp)
    radians = np.arctan2(tp[1]- sp[1], tp[0]- sp[0])- np.arctan2(fp[1]- sp[1], fp[0]- sp[0])
    angle = np.abs(radians *180/ np.pi)return angle if angle <=180else360- angle

# Function to get pose angles from an imagedefget_pose_angles(img):
    img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    results = pose.process(img_rgb)ifnot results.pose_landmarks:returnNone

    landmarks = results.pose_landmarks.landmark

    defget_coords(landmark):return[landmarks[landmark.value].x, landmarks[landmark.value].y]try:
        left_shoulder, left_elbow, left_wrist =map(get_coords,[mp_pose.PoseLandmark.LEFT_SHOULDER,
                                                                 mp_pose.PoseLandmark.LEFT_ELBOW,
                                                                 mp_pose.PoseLandmark.LEFT_WRIST])
        right_shoulder, right_elbow, right_wrist =map(get_coords,[mp_pose.PoseLandmark.RIGHT_SHOULDER,
                                                                    mp_pose.PoseLandmark.RIGHT_ELBOW,
                                                                    mp_pose.PoseLandmark.RIGHT_WRIST])
        left_hip, left_knee, left_ankle =map(get_coords,[mp_pose.PoseLandmark.LEFT_HIP,
                                                           mp_pose.PoseLandmark.LEFT_KNEE,
                                                           mp_pose.PoseLandmark.LEFT_ANKLE])
        right_hip, right_knee, right_ankle =map(get_coords,[mp_pose.PoseLandmark.RIGHT_HIP,
                                                              mp_pose.PoseLandmark.RIGHT_KNEE,
                                                              mp_pose.PoseLandmark.RIGHT_ANKLE])
        center_shoulder =[(left_shoulder[0]+ right_shoulder[0])/2,(left_shoulder[1]+ right_shoulder[1])/2]
        center_hip =[(left_hip[0]+ right_hip[0])/2,(left_hip[1]+ right_hip[1])/2]
        vertical_refp =[(left_hip[0]+ right_hip[0])/2,(left_hip[1]+ right_hip[1])/2+10]

        angles =[
            calculate_angle(left_shoulder, left_elbow, left_wrist),
            calculate_angle(right_shoulder, right_elbow, right_wrist),
            calculate_angle(left_hip, left_knee, left_ankle),
            calculate_angle(right_hip, right_knee, right_ankle),180- calculate_angle(center_shoulder, center_hip, vertical_refp)]return angles

    except Exception as e:print(f"Error in get_pose_angles: {e}")returnNone

其中的landmark.png是各个关节的相对位置和关系地图，从外部加载。

接着，我们通过计算与标准角度间的差距，用分段函数映射，对于绝对值在10以下，记为满分；对于10到70之间，采用平方衰减；对于70以上，采用指数衰减；具体参数通过反馈调参实现，代码实现如下：

        angles = np.round(angles,2)
        diff_angles = angles - np.array(standard_angles)
        frame_scores = np.zeros(len(diff_angles))for part inrange(len(diff_angles)):
            abs_diff =abs(diff_angles[part])if abs_diff <=10:
                frame_scores[part]=100-0.03* abs_diff * abs_diff
            elif10< abs_diff <=70:
                frame_scores[part]=94.23-0.02*(abs_diff -8)*(abs_diff -8)else:
                frame_scores[part]=17.35* math.exp(-(abs_diff -70)/20)

局限性

接入外围接口和设备之后，我们发现这个算法的评分十分不稳定，而且可信度不高。具体表现为：暂停视频同一帧比较时，评分仍然会较大波动；并且即便人坐着，标准视频站着，都有可能得到较高的分数。

解决方案

经过中途调试和分析并且将点位绘制在实时视频流上，我们发现即便人不动作，标记点位都会有较大的波动，表现为动作的位移和置信度的波动，一个方案是通过提高算法参数规模，但是我们暂时没有更多算力资源，OpenPose的后端每秒只能处理10帧，PosNet的20版本每秒可以处理20帧（相当糟糕的表现）。另一个更方便的方案是通过缓存，用最近的历史平滑掉波动的问题。
对于非波动性误差的产生，我经过反复验证，确定是由于时间误差产生的，因为上述的算法几乎完全无视了这个问题，因为我们的算力和人反应力是不稳定的，会产生动态的延迟，因此我们需要用一个更大的尺度来衡量这个分数标准，也是从另一个方面平滑了波动的效果。

第三版

第三版为个人研究所得，个人想法先通过划分时间和关节动态，得到各个动作的时间区间。

• 单位时间：通过时间戳，降低算力和录制、标准视频帧率对最终得分的影响
• 关节动态的定义：关节在单位时间内经过缓存平滑计算后，仍然和历史0保留位置存在大于阈值的距离，则标记为关节动态。
• 关节方向：以逆时针旋转，Ｘ＋为零弧度，计算从上一个位置到平滑后的当前位置的方向。
• 动作同一性：如果关节动态未改变，且关节运动方向与上一个状态差<30°
• 动作连续性：如果关节动态未改变，且关节方向运动上一个状态差大于30°晓宇45°；否则切分为另一个动作，将上一个动作记录为最小评分单位。

functionupdateJointState(pose, currentTime){// 最后一次的pose可能是undefine，因此需要判断一下if(!pose){return;}
    
    positions = pose.keypoints.map(keypoint=>({x: keypoint.position.x,y: keypoint.position.y }));if(bufferPositions.length ==0){pushBuffer(positions, currentTime, jointStates, jointSpeeds, lastDirections);return;}
    buffer =getBuffer();
    console.log(buffer);
    lastPositions = buffer.positions;
    prevJointStates = buffer.states;
    prevLastDirections = buffer.directions;
    lastTime = buffer.time;if(!lastTime){return;}const dt =(currentTime - lastTime);if(dt ===0){return;}// Check if all joints are moving in the same direction  let notSameDirection =0;let directionSum =0;// Calculate speed and angle for each joint  // 将jointStates清空for(let i =0; i <16; i++){  
        jointStates[i]=false;  
        jointSpeeds[i]=0;  
        lastDirections[i]=0;}for(let i =5; i <16; i++){if(pose.keypoints[i].score <0.5){continue}const dx = positions[i].x - lastPositions[i].x;const dy = positions[i].y - lastPositions[i].y;const speed = Math.sqrt(dx * dx + dy * dy)/ dt * pose_weight[i];// px/msconst angle = Math.atan2(dy, dx)*180/ Math.PI;// degree/ms thresh = 100degree/1000ms

        jointSpeeds[i]= speed;// 关节速度
        lastDirections[i]= angle;// 运动方向，x轴正方向为0°，逆时针为正方向 
        directionSum += angle;if(speed > threshold){  
            jointStates[i]=true;// 关节是否在运动}else{  
            jointStates[i]=false;}}// Check if all joints are moving in the same direction ，理想状态下所有，但是实际上只要大于8个就行const avgDirection = directionSum / jointStates.slice(5).filter(Boolean).length;  
    notSameDirection =0;for(let i =5; i <16; i++){if(jointStates[i]){if(Math.abs(lastDirections[i]- avgDirection)> same_direction_threshold){  
                notSameDirection +=1;break;}}}if(notSameDirection <4){  
        jointStates.fill(false);let lowestMoving =-1, maxY =0;for(let i =5; i <16; i++){// find the lowest joint that is moving  (y坐标最大)  if(jointStates[i]&& positions[i].y > maxY){  
                maxY = positions[i].y;  
                lowestMoving = i -5;}}if(lowestMoving !==-1){  
            jointStates[lowestMoving +5]=true;// Mark the other joints as not moving except for the lowest one  }}// Remove conflicting joints  （如果挥左手，左肩、左肘、左腕都会动，当左腕加权速度最大时，只需要告诉用户动左腕就行）for(let i =0; i < pose_conflict.length; i++){const joints = pose_conflict[i];  
        max_speed =0;  
        max_joint =-1;// Reset the moving state of conflicting joints  
        joints.forEach(j=> jointStates[j]=false);for(let j =0; j < joints.length; j++){if(jointSpeeds[joints[j]]> max_speed){  
                max_speed = jointSpeeds[joints[j]];  
                max_joint = j;}}if(max_joint !==-1){  
            jointStates[joints[max_joint]]=true;}}

    console.log(jointStates);
    console.log(jointSpeeds);   
    console.log(lastDirections);// Check continuity  let isThisContinuous =true;let minorChanged =false;let movingChanged =false;let directionChanged =false;for(let i =5; i <16; i++){if(jointStates[i]!= prevJointStates[i]){  
            isThisContinuous =false;  
            movingChanged =true;break;}if(jointStates[i]){if(Math.abs(lastDirections[i]- prevLastDirections[i])> same_direction_threshold){  
                isThisContinuous =false;
                directionChanged =true;break;}elseif(Math.abs(lastDirections[i]- prevLastDirections[i])> minor_changed_direction_threshold){// 改变方向小于45°但大于30°时，认为是连续的，记录进入history 
                minorChanged =true;break;}}}// 容错检查if(minorChanged){updateHistory(positions, lastDirections);drawHistory();}
    document.getElementById('test').textContent ='---';if(!isThisContinuous){  
        interruptionCount++;if(interruptionCount >= incontinuity_thresh && dt > incontinuity_time){if(movingChanged){
                document.getElementById('test').textContent ='动作变化';}elseif(directionChanged){
                document.getElementById('test').textContent ='方向变化';}updateHistory(lastPositions, lastDirections);// Save last known continuous positions  drawHistory();// 在canvas_mask上画出连续动作轨迹  
            history =[];updateHistory(lastPositions, lastDirections);// Save current positions as renewed continuous positions  
            interruptionCount =0;}}else{  
        interruptionCount =0;// Reset counter on continuity  // Update last positions and time  }pushBuffer(positions, currentTime, jointStates, jointSpeeds, lastDirections);}

以上评分过程还在开发调试中，目前暂未发现问题。