345 lines
16 KiB
Python
345 lines
16 KiB
Python
import time
|
||
from copy import deepcopy
|
||
|
||
import numpy as np
|
||
|
||
from MCamera.mp_algorithm import MediapipeAlgorithmPlugin
|
||
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 SitUp_Old_1(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.speak_state = 0
|
||
self.state = 0
|
||
self.direction = 0
|
||
self.is_storard = True
|
||
# 初始化
|
||
self.countdown = 120
|
||
# 动作持续帧数
|
||
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.had_done = False
|
||
self.sign = 0
|
||
# 目标朝向
|
||
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.start_time = time.time()
|
||
self.countdown_flag.set()
|
||
|
||
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.time_2 = 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.time_1 = 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.6 and hand_range_2 > shoulder_range * 1.6:
|
||
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 += 1
|
||
if self.time_1 > 3:
|
||
self.state = 1
|
||
self.time_1 = 0
|
||
if self.speak_state == 0:
|
||
self.speak_driver.add_speak('手离肩')
|
||
self.speak_state = 1
|
||
|
||
if self.count >= 1:
|
||
if shoulder_angle <= 25:
|
||
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
|
||
self.time_1 = 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.time_1 = 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)
|
||
range = self.detector.findRange(hip_2_x, hip_2_y, knee_2_x, knee_2_y) * 0.5
|
||
hand_range = self.detector.findRange(elbow_2_x, elbow_2_y, knee_2_x, knee_2_y)
|
||
if hand_range_1 > shoulder_range * 1.6 and hand_range_2 > shoulder_range * 1.6:
|
||
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 <= 25:
|
||
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 >= 68 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)))
|