276 lines
13 KiB
Python
276 lines
13 KiB
Python
import threading
|
||
import time
|
||
from copy import deepcopy
|
||
|
||
import numpy as np
|
||
|
||
from MCamera.mn_camera import MN_RESULT
|
||
from PureBackend.general import *
|
||
from MCamera.mn_algorithm import MoveNetAlgorithmPlugin
|
||
from MCamera.mp_camera import MP_RESULT
|
||
from MCamera.camera import *
|
||
from Speaker.speak_base import beep
|
||
from Database.manager_database import *
|
||
from score_doc import get_fin_score
|
||
from .base_exercise import BaseExercise
|
||
|
||
|
||
class PushUp_3(BaseExercise):
|
||
def __init__(self, info, statistic_time=120, camera=None):
|
||
super().__init__(info, statistic_time, camera=camera)
|
||
self.countdown_flag = threading.Event() # 计时线程
|
||
# 个数统计
|
||
self.bar = None
|
||
self.per = None
|
||
self.count = 0
|
||
# 当前状态
|
||
self.pre_pos = 0
|
||
self.state = 0
|
||
self.speak_state = 0
|
||
self.direction = 1
|
||
# 动作持续帧数
|
||
self.time = 0
|
||
self.last_time = 0
|
||
self.interval = 0.5
|
||
# 开始标志
|
||
self.form = 0
|
||
# 初始化
|
||
self.countdown = 120
|
||
self.shoulder_angle = 23
|
||
# 状态反馈
|
||
self.feedback = "开始"
|
||
self.had_done = False
|
||
self.sign = 0
|
||
self.time_1 = 0
|
||
self.time_2 = 0
|
||
self.dir = 0
|
||
self.exercise_type = "俯卧撑"
|
||
# 俯卧撑参数
|
||
self.corner = (180, 430, 80, 560)
|
||
self.config = (0.7, 50, 55, 60)
|
||
MoveNetAlgorithmPlugin.set_corner(corner=self.corner)
|
||
MoveNetAlgorithmPlugin.set_config(config=self.config)
|
||
|
||
def get_result(self):
|
||
# 人员年龄
|
||
age = self.info.get(AGE)
|
||
# 人员性别
|
||
gender = "woman" if self.info.get(GENDER) == "女" else "man"
|
||
count = self.count
|
||
score = get_fin_score.Military(gender, int(age),
|
||
integratedProjectResult=int(count)).IntegratedProjectScoreEvaluation()
|
||
if age < 40:
|
||
score = 0
|
||
result = {
|
||
"count": int(count),
|
||
"score": score,
|
||
"countdown": self.countdown,
|
||
"had_done": self.had_done
|
||
}
|
||
return result
|
||
|
||
def retail_counting(self):
|
||
retail_counting = 3
|
||
counting = 0
|
||
first_time = time.time()
|
||
while True:
|
||
this_time = time.time()
|
||
if this_time - first_time > counting:
|
||
if retail_counting > 0:
|
||
self.speak_driver.add_speak(f"{retail_counting}！")
|
||
self.speak_driver.wait_4_speak()
|
||
counting += 1
|
||
retail_counting -= 1
|
||
else:
|
||
break
|
||
threading.Thread(target=beep, daemon=True).start()
|
||
self.countdown_flag.set()
|
||
self.start_time = time.time()
|
||
|
||
def speak_counting(self, counting_times, name, age):
|
||
self.speak_driver.start()
|
||
self.speak_driver.speed_control(200)
|
||
self.speak_driver.volume_control(1)
|
||
self.speak_driver.add_speak("考试人员{}".format(name))
|
||
self.speak_driver.add_speak(f"考试项目{self.exercise_type}")
|
||
if age < 40:
|
||
self.speak_driver.add_speak(f"注意，该考生年龄小于40岁，{self.exercise_type}项目成绩无效")
|
||
|
||
def count_down(self):
|
||
seconds = 120
|
||
while True:
|
||
self.countdown_flag.wait()
|
||
this_time = time.time()
|
||
count_time = this_time - self.start_time
|
||
second = int(seconds - count_time)
|
||
self.countdown = f"{second}"
|
||
time.sleep(0.1)
|
||
if second == 0:
|
||
self.countdown_flag.clear()
|
||
break
|
||
|
||
def run(self) -> None:
|
||
self.speak_counting(5, self.info[NAME], self.info[AGE])
|
||
# 准备倒计时
|
||
threading.Thread(target=self.count_down).start()
|
||
self.is_start = True
|
||
# 不停更新计算和更新画面
|
||
threading.Thread(target=self.thread_counting_streaming, daemon=True).start()
|
||
while not self.is_done():
|
||
_img = deepcopy(self.cap.get_frame())
|
||
self.img = self.display(_img)
|
||
if self.countdown == 120:
|
||
self.speak_driver.add_speak(f"长时间未进入准备状态，考试结束")
|
||
elif int(self.countdown) <= 1:
|
||
self.had_done = True
|
||
self.speak_driver.add_speak(f"时间到，考试结束")
|
||
elif 1 < int(self.countdown) < 120:
|
||
self.speak_driver.add_speak(f"考试终止")
|
||
self.is_start = False
|
||
|
||
def display(self, img):
|
||
cv2.rectangle(img, (self.corner[2], self.corner[0]), (self.corner[3], self.corner[1]), (0, 255, 0), 2)
|
||
# 检查俯卧撑运动过程
|
||
if self.sign == 1 and self.dir == 1 and self.countdown_flag.is_set():
|
||
self.detector.drawPoint(img, LEFT_SHOULDER, LEFT_ELBOW, LEFT_ANKLE, bias_x=self.corner[2],
|
||
bias_y=self.corner[0])
|
||
if self.sign == 1 and self.dir == 2 and self.countdown_flag.is_set():
|
||
self.detector.drawPoint(img, RIGHT_SHOULDER, RIGHT_ELBOW, RIGHT_ANKLE, bias_x=self.corner[2],
|
||
bias_y=self.corner[0])
|
||
# 绘制状态条
|
||
if self.sign == 1:
|
||
cv2.rectangle(img, (650 - 100, 60), (650 - 80, 282), (255, 255, 255), 2)
|
||
if self.bar and self.per:
|
||
cv2.rectangle(img, (650 - 98, int(self.bar)), (650 - 82, 280), (102, 106, 233),
|
||
cv2.FILLED)
|
||
cv2.putText(img, f'{int(self.per)}%', (650 - 125, 320), cv2.FONT_HERSHEY_PLAIN, 2,
|
||
(255, 255, 255), 2)
|
||
|
||
# 绘制计数器
|
||
cv2.rectangle(img, (0, 480 - 120), (120, 480), (102, 106, 233), cv2.FILLED)
|
||
cv2.putText(img, str(int(self.count)), (10, 480 - 35), cv2.FONT_HERSHEY_PLAIN, 5,
|
||
(255, 255, 255), 5)
|
||
|
||
# 展示状态反馈
|
||
cv2.rectangle(img, (640 - 160, 0), (640, 50), (102, 106, 233), cv2.FILLED)
|
||
img = self.cv2_img_add_text(img, self.feedback, 640 - 120, 5, (255, 255, 255), 38)
|
||
|
||
return img
|
||
|
||
def analysis(self, frame):
|
||
catch_time = frame[CATCH_TIME]
|
||
if catch_time < self.start_cal_time:
|
||
return
|
||
img = frame[FRAME_DAT]
|
||
lm_list = frame[MN_RESULT]
|
||
self.detector.set_result(lm_list)
|
||
|
||
if len(lm_list) != 0:
|
||
shoulder_1_x = self.detector.findPosition(img, False)[LEFT_SHOULDER]['key_points'][0]
|
||
shoulder_2_x = self.detector.findPosition(img, False)[RIGHT_SHOULDER]['key_points'][0]
|
||
shoulder_1_y = self.detector.findPosition(img, False)[LEFT_SHOULDER]['key_points'][1]
|
||
shoulder_2_y = self.detector.findPosition(img, False)[RIGHT_SHOULDER]['key_points'][1]
|
||
vis_elbow_1 = self.detector.findPosition(img, False)[LEFT_ELBOW]['confidence']
|
||
vis_elbow_2 = self.detector.findPosition(img, False)[RIGHT_ELBOW]['confidence']
|
||
vis_knee_1 = self.detector.findPosition(img, False)[LEFT_KNEE]['confidence']
|
||
vis_knee_2 = self.detector.findPosition(img, False)[RIGHT_KNEE]['confidence']
|
||
Concave_angle_1 = self.detector.findConcave(LEFT_SHOULDER, LEFT_HIP, LEFT_KNEE)
|
||
Concave_angle_2 = self.detector.findConcave(RIGHT_SHOULDER, RIGHT_HIP, RIGHT_KNEE)
|
||
knee_1_x = self.detector.findPosition(img, False)[LEFT_KNEE]['key_points'][0]
|
||
knee_2_x = self.detector.findPosition(img, False)[RIGHT_KNEE]['key_points'][0]
|
||
knee_1_y = self.detector.findPosition(img, False)[LEFT_KNEE]['key_points'][1]
|
||
knee_2_y = self.detector.findPosition(img, False)[RIGHT_KNEE]['key_points'][1]
|
||
ankle_1_x = self.detector.findPosition(img, False)[LEFT_ANKLE]['key_points'][0]
|
||
ankle_2_x = self.detector.findPosition(img, False)[RIGHT_ANKLE]['key_points'][0]
|
||
ankle_1_y = self.detector.findPosition(img, False)[LEFT_ANKLE]['key_points'][1]
|
||
ankle_2_y = self.detector.findPosition(img, False)[RIGHT_ANKLE]['key_points'][1]
|
||
elbow_1 = self.detector.findAngle(img, LEFT_SHOULDER, LEFT_ELBOW, LEFT_WRIST, False)
|
||
elbow_2 = self.detector.findAngle(img, RIGHT_SHOULDER, RIGHT_ELBOW, RIGHT_WRIST, False)
|
||
|
||
if self.pre_pos == 0:
|
||
self.pre_pos = 1
|
||
self.speak_driver.add_speak("请进入准备状态")
|
||
|
||
if self.form == 0:
|
||
if ankle_1_y > shoulder_1_y or ankle_2_y > shoulder_2_y:
|
||
if self.pre_pos == 1:
|
||
if vis_elbow_1 > 0.8 and vis_elbow_1 > vis_elbow_2 and vis_knee_1 > 0.8:
|
||
self.time_1 += 1
|
||
if self.time_1 > 4:
|
||
self.dir = 1
|
||
self.sign = 1
|
||
self.time_1 = 0
|
||
self.form = 1
|
||
elif vis_elbow_2 > 0.8 and vis_elbow_1 < vis_elbow_2 and vis_knee_2 > 0.8:
|
||
self.time_2 += 1
|
||
if self.time_2 > 4:
|
||
self.dir = 2
|
||
self.sign = 1
|
||
self.time_2 = 0
|
||
self.form = 1
|
||
if self.sign == 1:
|
||
self.speak_driver.add_speak("准备开始考试")
|
||
threading.Thread(target=self.retail_counting).start()
|
||
|
||
# 检查俯卧撑运动过程
|
||
if self.sign == 1 and self.dir == 1 and self.countdown_flag.is_set():
|
||
shoulder_angle = self.detector.findIncludedAngle(ankle_1_x, ankle_1_y, shoulder_1_x, shoulder_1_y)
|
||
hip_angle = self.detector.findIncludedAngle(ankle_1_x, ankle_1_y, knee_1_x, knee_1_y)
|
||
# 成功概率
|
||
self.per = np.interp(elbow_1, (90, 150), (100, 0))
|
||
# 显示进度栏
|
||
self.bar = np.interp(elbow_1, (90, 150), (62, 280))
|
||
if hip_angle < 70:
|
||
if elbow_1 > 140 and 150 <= Concave_angle_1 <= 210:
|
||
self.feedback = "下落"
|
||
if self.direction == 0:
|
||
self.time += 1
|
||
if self.time > 1:
|
||
self.count += 0.5
|
||
self.direction = 1
|
||
self.time = 0
|
||
self.speak_state = 0
|
||
self.shoulder_angle = shoulder_angle
|
||
if self.count % 1 == 0:
|
||
self.speak_driver.speed_control(200)
|
||
self.speak_driver.add_speak("{}".format(int(self.count)))
|
||
if self.shoulder_angle - shoulder_angle < 3 and elbow_1 < 115 and time.time() - self.last_time > self.interval:
|
||
self.feedback = "上升"
|
||
if self.direction == 1:
|
||
self.time += 1
|
||
if self.time > 1:
|
||
self.count += 0.5
|
||
self.direction = 0
|
||
self.time = 0
|
||
self.last_time = time.time()
|
||
|
||
if self.sign == 1 and self.dir == 2 and self.countdown_flag.is_set():
|
||
shoulder_angle = self.detector.findIncludedAngle(ankle_2_x, ankle_2_y, shoulder_2_x, shoulder_2_y)
|
||
hip_angle = self.detector.findIncludedAngle(ankle_2_x, ankle_2_y, knee_2_x, knee_2_y)
|
||
# 成功概率
|
||
self.per = np.interp(elbow_2, (90, 150), (100, 0))
|
||
# 显示进度栏
|
||
self.bar = np.interp(elbow_2, (90, 150), (62, 280))
|
||
if hip_angle < 70:
|
||
if elbow_2 > 140 and 150 <= Concave_angle_2 <= 210:
|
||
self.feedback = "下落"
|
||
if self.direction == 0:
|
||
self.time += 1
|
||
if self.time > 1:
|
||
self.count += 0.5
|
||
self.direction = 1
|
||
self.time = 0
|
||
self.speak_state = 0
|
||
self.shoulder_angle = shoulder_angle
|
||
if self.count % 1 == 0:
|
||
self.speak_driver.speed_control(200)
|
||
self.speak_driver.add_speak("{}".format(int(self.count)))
|
||
if self.shoulder_angle - shoulder_angle < 3 and elbow_1 < 115 and time.time() - self.last_time > self.interval:
|
||
self.feedback = "上升"
|
||
if self.direction == 1:
|
||
self.time += 1
|
||
if self.time > 1:
|
||
self.count += 0.5
|
||
self.direction = 0
|
||
self.time = 0
|
||
self.last_time = time.time() |