[e8481a]: / src / pipeline / pose_engine.py

Download this file

196 lines (150 with data), 6.4 kB

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
from src import DEFAULT_DATA_DIR
import logging
import time
from PIL import ImageDraw
from pathlib import Path
from src.pipeline.posenet_model import Posenet_MobileNet
from src.pipeline.movenet_model import Movenet
log = logging.getLogger(__name__)
KEYPOINTS = (
'nose',
'left eye',
'right eye',
'left ear',
'right ear',
'left shoulder',
'right shoulder',
'left elbow',
'right elbow',
'left wrist',
'right wrist',
'left hip',
'right hip',
'left knee',
'right knee',
'left ankle',
'right ankle'
)
class Keypoint:
__slots__ = ['k', 'yx', 'score']
def __init__(self, k, yx, score=None):
self.k = k
self.yx = yx
self.score = score
def __repr__(self):
return 'Keypoint(<{}>, {}, {})'.format(self.k, self.yx, self.score)
class Pose:
__slots__ = ['keypoints', 'score']
def __init__(self, keypoints, score=None):
assert len(keypoints) == len(KEYPOINTS)
self.keypoints = keypoints
self.score = score
def __repr__(self):
return 'Pose({}, {})'.format(self.keypoints, self.score)
class PoseEngine():
"""Engine used for pose tasks."""
def __init__(self, tfengine=None, model_name=None, context=None):
"""Creates a PoseEngine wrapper around an initialized tfengine.
"""
assert tfengine is not None
assert model_name is not None
if model_name == 'movenet':
self._model = Movenet(tfengine)
elif model_name == 'mobilenet':
self._model = Posenet_MobileNet(tfengine)
if context:
self._sys_data_dir = context.data_dir
else:
self._sys_data_dir = DEFAULT_DATA_DIR
self._sys_data_dir = Path(self._sys_data_dir)
self.confidence_threshold = self._model.confidence_threshold
self._tensor_image_height = self._model._tensor_image_height
self._tensor_image_width = self._model._tensor_image_width
def draw_kps(self, kps, template_image):
pil_im = template_image
draw = ImageDraw.Draw(pil_im)
leftShoulder = False
rightShoulder = False
scoreList = {'LShoulder_score':0,'RShoulder_score':0,'LHip_score':0,'RHip_score':0}
for i in range(kps.shape[0]):
x, y, r = int(round(kps[i, 1])), int(round(kps[i, 0])), 1
if i == 5:
leftShoulder = True
leftShoulder_point = [x, y]
scoreList['LShoulder_score'] = kps[i,-1]
if i == 6:
rightShoulder = True
rightShoulder_point = [x, y]
scoreList['RShoulder_score'] = kps[i,-1]
leftUpPoint = (x-r, y-r)
rightDownPoint = (x+r, y+r)
twoPointList = [leftUpPoint, rightDownPoint]
draw.ellipse(twoPointList, fill=(0, 255, 0, 255))
if i == 11 and leftShoulder:
leftHip_point = [x, y]
scoreList['LHip_score'] = kps[i,-1]
draw.line((leftShoulder_point[0],leftShoulder_point[1], leftHip_point[0],leftHip_point[1]), fill='green', width=3)
if i == 12 and rightShoulder:
rightHip_point = [x, y]
scoreList['RHip_score'] = kps[i,-1]
draw.line((rightShoulder_point[0],rightShoulder_point[1], rightHip_point[0],rightHip_point[1]), fill='green', width=3)
return pil_im, scoreList
def get_result(self, img):
kps, template_image, thumbnail, _inference_time = self._model.execute_model(img)
output_img, scoreList = self.draw_kps(kps, template_image)
return thumbnail, output_img, scoreList, _inference_time
def detect_poses(self, img):
"""
Detects poses in a given image.
:Parameters:
----------
img : PIL.Image
Input Image for AI model detection.
:Returns:
-------
poses:
A list of Pose objects with keypoints and confidence scores
PIL.Image
Resized image fitting the AI model input tensor.
"""
kps, template_image, thumbnail, _ = self._model.execute_model(img)
poses = []
keypoint_dict = {}
cnt = 0
keypoint_count = kps.shape[0]
for point_i in range(keypoint_count):
x, y = kps[point_i, 1], kps[point_i, 0]
prob = kps[point_i, 2]
if prob > self.confidence_threshold and \
0 < y < self._tensor_image_height and \
0 < x < self._tensor_image_width:
cnt += 1
if log.getEffectiveLevel() <= logging.DEBUG:
# development mode
# draw on image and save it for debugging
draw = ImageDraw.Draw(template_image)
draw.line(((0, 0), (x, y)), fill='blue')
else:
prob = 0
keypoint = Keypoint(KEYPOINTS[point_i], [x, y], prob)
keypoint_dict[KEYPOINTS[point_i]] = keypoint
# overall pose score is calculated as the average of all
# individual keypoint scores
pose_score = cnt/keypoint_count
log.debug(f"Overall pose score (keypoint score average): {pose_score}")
poses.append(Pose(keypoint_dict, pose_score))
if cnt > 0 and log.getEffectiveLevel() <= logging.DEBUG:
# development mode
# save template_image for debugging
timestr = int(time.monotonic()*1000)
log.debug(f"Detected a pose with {cnt} keypoints that score over \
the minimum confidence threshold of \
{self.confidence_threshold}.")
debug_image_file_name = \
f'tmp-pose-detect-image-time-{timestr}-keypoints-{cnt}.jpg'
# template_image.save(
# Path(self._sys_data_dir,
# debug_image_file_name),
# format='JPEG')
log.debug(f"Debug image saved: {debug_image_file_name}")
return poses, thumbnail, pose_score