344 lines
16 KiB
Python
344 lines
16 KiB
Python
|
import time
|
|||
|
|
|
|||
|
|
from MCamera.mp_algorithm import MediapipeAlgorithmPlugin
|
|||
|
|
from MCamera.mp_camera import MP_RESULT
|
|||
|
|
from MCamera.camera import *
|
|||
|
|
|
|||
|
|
import cv2
|
|||
|
|
import numpy as np
|
|||
|
|
from score_doc import get_fin_score
|
|||
|
|
from Speaker.speak_base import beep
|
|||
|
|
from copy import deepcopy
|
|||
|
|
from Database.manager_database import *
|
|||
|
|
from .base_exercise import BaseExercise
|
|||
|
|
import pyttsx3
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SitUp_Old_2(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.countdown = 120
|
|||
|
|
self.pre_pos = 0
|
|||
|
|
self.speak_state = 0
|
|||
|
|
self.state = 0
|
|||
|
|
self.direction = 0
|
|||
|
|
self.is_storard = True
|
|||
|
|
# 初始化
|
|||
|
|
# 动作持续帧数
|
|||
|
|
self.time = 0
|
|||
|
|
self.time_1 = 0
|
|||
|
|
self.time_2 = 0
|
|||
|
|
self.last_time = 0
|
|||
|
|
self.interval = 0.5
|
|||
|
|
# 开始标志
|
|||
|
|
self.form = 0
|
|||
|
|
# 状态反馈
|
|||
|
|
self.feedback = "开始"
|
|||
|
|
self.sign = 0
|
|||
|
|
self.had_done = False
|
|||
|
|
# 目标朝向
|
|||
|
|
self.dir = 0
|
|||
|
|
self.exercise_type = "仰卧起坐"
|
|||
|
|
# 仰卧起坐参数
|
|||
|
|
self.corner = (160, 460, 100, 520)
|
|||
|
|
self.config = (True, 1, True, False, True, 0.75, 0.8)
|
|||
|
|
MediapipeAlgorithmPlugin.set_corner(corner=self.corner)
|
|||
|
|
MediapipeAlgorithmPlugin.set_config(config=self.config)
|
|||
|
|
|
|||
|
|
def get_result(self):
|
|||
|
|
# 人员年龄
|
|||
|
|
age = self.info.get(AGE)
|
|||
|
|
# 人员性别
|
|||
|
|
gender = "woman" if self.info.get(GENDER) == "女" else "man"
|
|||
|
|
score = get_fin_score.Military(gender, int(age), sitUpsResult=int(self.count)).SitUpsScoreEvaluation()
|
|||
|
|
result = {
|
|||
|
|
"count": int(self.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):
|
|||
|
|
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}")
|
|||
|
|
|
|||
|
|
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])
|
|||
|
|
# 准备倒计时
|
|||
|
|
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.form == 0:
|
|||
|
|
self.detector.draw_dashed_line(img, 100, 400, 520, 440, (0, 0, 255))
|
|||
|
|
if self.dir == 1 and self.sign == 1 and self.countdown_flag.is_set():
|
|||
|
|
self.detector.drawPoint_more(img, [11, 13, 16, 23, 25, 27], bias_x=self.corner[2], bias_y=self.corner[0])
|
|||
|
|
elif self.dir == 2 and self.sign == 1 and self.countdown_flag.is_set():
|
|||
|
|
self.detector.drawPoint_more(img, [12, 14, 15, 24, 26, 28], 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[MP_RESULT]
|
|||
|
|
self.detector.set_result(lm_list)
|
|||
|
|
|
|||
|
|
if self.pre_pos == 0:
|
|||
|
|
self.pre_pos = 1
|
|||
|
|
self.speak_driver.add_speak("请进入准备状态")
|
|||
|
|
|
|||
|
|
if len(lm_list) != 0:
|
|||
|
|
eye_1_y = self.detector.findPosition(img, False)[3][2]
|
|||
|
|
eye_2_y = self.detector.findPosition(img, False)[6][2]
|
|||
|
|
shoulder_1_x = self.detector.findPosition(img, False)[11][1]
|
|||
|
|
shoulder_2_x = self.detector.findPosition(img, False)[11][1]
|
|||
|
|
shoulder_1_y = self.detector.findPosition(img, False)[11][2]
|
|||
|
|
shoulder_2_y = self.detector.findPosition(img, False)[11][2]
|
|||
|
|
elbow_1_x = self.detector.findPosition(img, False)[13][1]
|
|||
|
|
elbow_1_y = self.detector.findPosition(img, False)[13][2]
|
|||
|
|
elbow_2_x = self.detector.findPosition(img, False)[14][1]
|
|||
|
|
elbow_2_y = self.detector.findPosition(img, False)[14][2]
|
|||
|
|
wrist_1_x = self.detector.findPosition(img, False)[15][1]
|
|||
|
|
wrist_2_x = self.detector.findPosition(img, False)[16][1]
|
|||
|
|
wrist_1_y = self.detector.findPosition(img, False)[15][2]
|
|||
|
|
wrist_2_y = self.detector.findPosition(img, False)[16][2]
|
|||
|
|
hip_1_x = self.detector.findPosition(img, False)[23][1]
|
|||
|
|
hip_2_x = self.detector.findPosition(img, False)[24][1]
|
|||
|
|
hip_1_y = self.detector.findPosition(img, False)[23][2]
|
|||
|
|
hip_2_y = self.detector.findPosition(img, False)[24][2]
|
|||
|
|
knee_1_x = self.detector.findPosition(img, False)[25][1]
|
|||
|
|
knee_2_x = self.detector.findPosition(img, False)[26][1]
|
|||
|
|
knee_1_y = self.detector.findPosition(img, False)[25][2]
|
|||
|
|
knee_2_y = self.detector.findPosition(img, False)[26][2]
|
|||
|
|
vis_hip_1 = self.detector.findPosition(img, False)[25][4]
|
|||
|
|
vis_hip_2 = self.detector.findPosition(img, False)[26][4]
|
|||
|
|
ankle_1_x = self.detector.findPosition(img, False)[27][1]
|
|||
|
|
ankle_2_x = self.detector.findPosition(img, False)[28][1]
|
|||
|
|
ankle_1_y = self.detector.findPosition(img, False)[27][2]
|
|||
|
|
ankle_2_y = self.detector.findPosition(img, False)[28][2]
|
|||
|
|
|
|||
|
|
if self.pre_pos == 0:
|
|||
|
|
self.pre_pos = 1
|
|||
|
|
self.speak_driver.add_speak("请进入准备状态")
|
|||
|
|
|
|||
|
|
if self.form == 0:
|
|||
|
|
angle_1 = self.detector.findIncludedAngle(hip_1_x, hip_1_y, ankle_1_x, ankle_1_y)
|
|||
|
|
angle_2 = self.detector.findIncludedAngle(hip_2_x, hip_2_y, ankle_2_x, ankle_2_y)
|
|||
|
|
if angle_1 < 40 and angle_2 < 40 and eye_1_y > knee_1_y and eye_2_y > knee_2_y:
|
|||
|
|
if self.pre_pos == 1:
|
|||
|
|
if vis_hip_1 > 0.8 and vis_hip_1 > vis_hip_2:
|
|||
|
|
self.time_1 += 1
|
|||
|
|
if self.time_1 > 3:
|
|||
|
|
self.dir = 1
|
|||
|
|
self.sign = 1
|
|||
|
|
self.time_1 = 0
|
|||
|
|
self.form = 1
|
|||
|
|
|
|||
|
|
elif vis_hip_2 > 0.8 and vis_hip_1 < vis_hip_2:
|
|||
|
|
self.time_2 += 1
|
|||
|
|
if self.time_2 > 3:
|
|||
|
|
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.dir == 1 and self.sign == 1 and self.countdown_flag.is_set():
|
|||
|
|
shoulder_angle = self.detector.findIncludedAngle(hip_1_x, hip_1_y, shoulder_1_x, shoulder_1_y)
|
|||
|
|
hand_range_1 = self.detector.findRange(wrist_1_x, wrist_1_y, shoulder_2_x, shoulder_2_y)
|
|||
|
|
hand_range_2 = self.detector.findRange(wrist_2_x, wrist_2_y, shoulder_1_x, shoulder_1_y)
|
|||
|
|
shoulder_range = self.detector.findRange(elbow_1_x, elbow_1_y, shoulder_1_x, shoulder_1_y)
|
|||
|
|
ankle_angle = self.detector.findIncludedAngle(ankle_1_x, ankle_1_y, hip_1_x, hip_1_y)
|
|||
|
|
hand_range = self.detector.findRange(elbow_1_x, elbow_1_y, knee_1_x, knee_1_y)
|
|||
|
|
range = self.detector.findRange(hip_1_x, hip_1_y, knee_1_x, knee_1_y) * 0.5
|
|||
|
|
|
|||
|
|
if hand_range_1 > shoulder_range * 1.7 and hand_range_2 > shoulder_range * 1.7:
|
|||
|
|
self.is_storard = False
|
|||
|
|
else:
|
|||
|
|
self.is_storard = True
|
|||
|
|
|
|||
|
|
self.per = np.interp(shoulder_angle, (25, 65), (0, 100))
|
|||
|
|
self.bar = np.interp(shoulder_angle, (25, 65), (62, 280))
|
|||
|
|
if ankle_angle < 70:
|
|||
|
|
if not self.is_storard:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 3:
|
|||
|
|
self.state = 1
|
|||
|
|
self.time = 0
|
|||
|
|
if self.speak_state == 0:
|
|||
|
|
self.speak_driver.add_speak('手离肩')
|
|||
|
|
self.speak_state = 1
|
|||
|
|
if self.count >= 1:
|
|||
|
|
if shoulder_angle <= 30:
|
|||
|
|
self.feedback = "上升"
|
|||
|
|
self.state = 0
|
|||
|
|
if self.direction == 0:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 2:
|
|||
|
|
self.count += 0.5
|
|||
|
|
self.direction = 1
|
|||
|
|
self.time = 0
|
|||
|
|
self.speak_state = 0
|
|||
|
|
else:
|
|||
|
|
self.feedback = "上升"
|
|||
|
|
self.state = 0
|
|||
|
|
if self.direction == 0:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 2:
|
|||
|
|
self.count += 0.5
|
|||
|
|
self.direction = 1
|
|||
|
|
self.time = 0
|
|||
|
|
self.speak_state = 0
|
|||
|
|
|
|||
|
|
if shoulder_angle >= 65 and hand_range < range and time.time() - self.last_time > self.interval and self.is_storard:
|
|||
|
|
self.feedback = "下落"
|
|||
|
|
if self.state == 0:
|
|||
|
|
if self.direction == 1:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 2:
|
|||
|
|
self.count += 0.5
|
|||
|
|
self.direction = 0
|
|||
|
|
self.time = 0
|
|||
|
|
self.last_time = time.time()
|
|||
|
|
if self.count % 1 == 0:
|
|||
|
|
self.speak_driver.add_speak("{}".format(int(self.count)))
|
|||
|
|
|
|||
|
|
elif self.dir == 2 and self.sign == 1 and self.countdown_flag.is_set():
|
|||
|
|
# 计算各点与腰点角度
|
|||
|
|
shoulder_angle = self.detector.findIncludedAngle(hip_2_x, hip_2_y, shoulder_2_x, shoulder_2_y)
|
|||
|
|
hand_range_1 = self.detector.findRange(wrist_2_x, wrist_2_y, shoulder_1_x, shoulder_1_y)
|
|||
|
|
hand_range_2 = self.detector.findRange(wrist_1_x, wrist_1_y, shoulder_2_x, shoulder_2_y)
|
|||
|
|
shoulder_range = self.detector.findRange(elbow_2_x, elbow_2_y, shoulder_2_x, shoulder_2_y)
|
|||
|
|
ankle_angle = self.detector.findIncludedAngle(ankle_2_x, ankle_2_y, hip_2_x, hip_2_y)
|
|||
|
|
hand_range = self.detector.findRange(elbow_2_x, elbow_2_y, knee_2_x, knee_2_y)
|
|||
|
|
range = self.detector.findRange(hip_2_x, hip_2_y, knee_2_x, knee_2_y) * 0.5
|
|||
|
|
|
|||
|
|
if hand_range_1 > shoulder_range * 1.7 and hand_range_2 > shoulder_range * 1.7:
|
|||
|
|
self.is_storard = False
|
|||
|
|
else:
|
|||
|
|
self.is_storard = True
|
|||
|
|
|
|||
|
|
# 展示进度栏
|
|||
|
|
self.per = np.interp(shoulder_angle, (30, 65), (0, 100))
|
|||
|
|
self.bar = np.interp(shoulder_angle, (30, 65), (62, 280))
|
|||
|
|
if ankle_angle < 70:
|
|||
|
|
if not self.is_storard:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 3:
|
|||
|
|
self.state = 1
|
|||
|
|
self.time = 0
|
|||
|
|
if self.speak_state == 0:
|
|||
|
|
self.speak_driver.add_speak('手离肩')
|
|||
|
|
self.speak_state = 1
|
|||
|
|
|
|||
|
|
if self.count >= 1:
|
|||
|
|
if shoulder_angle <= 30:
|
|||
|
|
self.feedback = "上升"
|
|||
|
|
self.state = 0
|
|||
|
|
if self.direction == 0:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 2:
|
|||
|
|
self.count += 0.5
|
|||
|
|
self.direction = 1
|
|||
|
|
self.time = 0
|
|||
|
|
self.speak_state = 0
|
|||
|
|
else:
|
|||
|
|
self.feedback = "上升"
|
|||
|
|
self.state = 0
|
|||
|
|
if self.direction == 0:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 2:
|
|||
|
|
self.count += 0.5
|
|||
|
|
self.direction = 1
|
|||
|
|
self.time = 0
|
|||
|
|
self.speak_state = 0
|
|||
|
|
|
|||
|
|
if shoulder_angle >= 65 and hand_range < range and time.time() - self.last_time > self.interval and self.is_storard:
|
|||
|
|
self.feedback = "下落"
|
|||
|
|
if self.state == 0:
|
|||
|
|
if self.direction == 1:
|
|||
|
|
self.time += 1
|
|||
|
|
if self.time > 2:
|
|||
|
|
self.count += 0.5
|
|||
|
|
self.direction = 0
|
|||
|
|
self.time = 0
|
|||
|
|
self.last_time = time.time()
|
|||
|
|
if self.count % 1 == 0:
|
|||
|
|
engine = pyttsx3.init()
|
|||
|
|
engine.setProperty('rate', 400)
|
|||
|
|
self.speak_driver.add_speak("{}".format(int(self.count)))
|