-
Notifications
You must be signed in to change notification settings - Fork 0
/
pose.py
155 lines (135 loc) · 4.34 KB
/
pose.py
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
import json
import socket
import cv2
import mediapipe as mp
mp_drawing = mp.solutions.drawing_utils
mp_pose = mp.solutions.pose
SOCKET_UDP_DEST_IP = "127.0.0.1"
SOCKET_UDP_DESK_PORT = 8885
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# For static images:
IMAGE_FILES = []
BG_COLOR = (192, 192, 192) # gray
LMS_KEY_PAIR = [
"NOSE",
"LEFT_EYE_INNER",
"LEFT_EYE",
"LEFT_EYE_OUTER",
"RIGHT_EYE_INNER",
"RIGHT_EYE",
"RIGHT_EYE_OUTER",
"LEFT_EAR",
"RIGHT_EAR",
"MOUTH_LEFT",
"MOUTH_RIGHT",
"LEFT_SHOULDER",
"RIGHT_SHOULDER",
"LEFT_ELBOW",
"RIGHT_ELBOW",
"LEFT_WRIST",
"RIGHT_WRIST",
"LEFT_PINKY",
"RIGHT_PINKY",
"LEFT_INDEX",
"RIGHT_INDEX",
"LEFT_THUMB",
"RIGHT_THUMB",
"LEFT_HIP",
"RIGHT_HIP",
"LEFT_KNEE",
"RIGHT_KNEE",
"LEFT_ANKLE",
"RIGHT_ANKLE",
"LEFT_HEEL",
"RIGHT_HEEL",
"LEFT_FOOT_INDEX",
"RIGHT_FOOT_INDEX"
]
# await websocket.send("websocket connected")
with mp_pose.Pose(
static_image_mode=True,
# model_complexity=2,
# enable_segmentation=True,
min_detection_confidence=0.5) as pose:
for idx, file in enumerate(IMAGE_FILES):
image = cv2.imread(file)
image_height, image_width, _ = image.shape
# Convert the BGR image to RGB before processing.
results = pose.process(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
if not results.pose_landmarks:
continue
print(
f'Nose coordinates: ('
f'{results.pose_landmarks.landmark[mp_pose.PoseLandmark.NOSE].x * image_width}, '
f'{results.pose_landmarks.landmark[mp_pose.PoseLandmark.NOSE].y * image_height})'
)
annotated_image = image.copy()
# Draw segmentation on the image.
# To improve segmentation around boundaries, consider applying a joint
# bilateral filter to "results.segmentation_mask" with "image".
condition = np.stack((results.segmentation_mask,) * 3, axis=-1) > 0.1
bg_image = np.zeros(image.shape, dtype=np.uint8)
bg_image[:] = BG_COLOR
annotated_image = np.where(condition, annotated_image, bg_image)
# Draw pose landmarks on the image.
mp_drawing.draw_landmarks(
annotated_image,
results.pose_landmarks,
mp_pose.POSE_CONNECTIONS,
# landmark_drawing_spec=mp_drawing_styles.get_default_pose_landmarks_style()
)
cv2.imwrite('/tmp/annotated_image' + str(idx) + '.png', annotated_image)
# Plot pose world landmarks.
mp_drawing.plot_landmarks(
results.pose_world_landmarks, mp_pose.POSE_CONNECTIONS)
# For webcam input:
cap = cv2.VideoCapture(0)
with mp_pose.Pose(
min_detection_confidence=0.5,
min_tracking_confidence=0.5) as pose: # pose 모듈 안의 클래스 Pose
n_frame = 0
while cap.isOpened():
success, image = cap.read()
if not success:
print("Ignoring empty camera frame.")
# If loading a video, use 'break' instead of 'continue'.
continue
n_frame += 1
# To improve performance, optionally mark the image as not writeable to
# pass by reference.
image.flags.writeable = False
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
results = pose.process(image)
# Draw the pose annotation on the image.
image.flags.writeable = True
image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
mp_drawing.draw_landmarks(
image,
results.pose_landmarks,
mp_pose.POSE_CONNECTIONS,
# landmark_drawing_spec=mp_drawing_styles.get_default_pose_landmarks_style()
)
# Flip the image horizontally for a selfie-view display.
cv2.imshow('MediaPipe Pose', cv2.flip(image, 1))
if (n_frame % 1 == 0):
# print(f"Frame: {n_frame}")
if results.pose_landmarks:
i_hate_python = {}
i_hate_python['LMs'] = {}
idx = 0
for element in results.pose_landmarks.landmark:
i_hate_python['LMs'][LMS_KEY_PAIR[idx]] = {
'x': element.x,
'y': element.y,
'z': element.z,
'visibility': element.visibility
}
idx += 1
# Serializing JSON
# json_object = json.dumps(i_hate_python, indent=4)
# print(json_object)
udp_socket.sendto(json.dumps(i_hate_python).encode(),
(SOCKET_UDP_DEST_IP, SOCKET_UDP_DESK_PORT))
if cv2.waitKey(5) & 0xFF == 27:
break
cap.release()