Context:
I am trying to process an image stream with a gesture recognition algorithm using ROS, OpenCV and Python. All the examples I've seen use OpenCV .cap directly from a webcam to process video, whereas I am inserting an image stream with a known image that updates at around 40fps.
Using a typical ROS2 image subscriber node, I convert a compressed image message that I'm sending to the device via a camera.
Then, I pass through the compressed image as an argument into my gesture recognition script. Once I do this, my code seems to get 'stuck' and the image data goes dark. I suspect it's because I've modified the while loop within my gesture recognition to try to accommodate new frames, except I know no way of reading in new frames to the gesture recognizer.
Please could anyone suggest any means of achieving my goal? (passing the compressed_img_msg straight into opencv video processing and getting frame by frame updates still without using build in video functionality)?
See code for ROS2 image node and gesture recognition main below:
My ROS2 compressed_image node is below:
////////////////////////////////////////////////////////////////////////////////////////////
compressed_image node
////////////////////////////////////////////////////////////////////////////////////////////
class CompressedImageSubscriber(Node):
def __init__(self, topic):
super().__init__("image_view_sub_node")
qos_policy = rclpy.qos.QoSProfile(reliability=rclpy.qos.ReliabilityPolicy.BEST_EFFORT,
history=rclpy.qos.HistoryPolicy.KEEP_LAST,
depth=1)
self.sub = self.create_subscription(CompressedImage, topic,
self.subscriber_callback, qos_profile=qos_policy)
self.received_msg = False
def subscriber_callback(self, compressed_image_msg):
if self.received_msg == False:
print("Image Data Received... Displaying Image in CV2 window")
self.received_msg = True
#this is the unprocessed image decoded from Unity
subscribed_image = CvBridge().compressed_imgmsg_to_cv2(compressed_image_msg, desired_encoding="bgr8")
cv2.imshow("CompressedImage", subscribed_image)
#pass the unprocessed image into the gesture recognition app
cv2.waitKey(1)
gesture_app.main(subscribed_image)
cv2.waitKey(1)
#call the gesture script, pass throuigh image as an argument here
#main code
def main():
rclpy.init()
topic_name = input("what is the topic name?: ")
my_sub = CompressedImageSubscriber(topic_name)
print("Waiting for data to be publisher over topic...")
try:
rclpy.spin(my_sub)
except KeyboardInterrupt:
my_sub.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
//////////////////////////////////////////////////////////////////////////////////////////
As you can see, once the image is decoded I pass through it to the gesture algorithm as an argument. The main script for this can be seen below:
/////////////////////////////////////////////////////////////////////////////////////////
GESTURE RECOGNITION MAIN FUNCTION
(source code: https://github.com/kinivi/hand-gesture-recognition-mediapipe)
/////////////////////////////////////////////////////////////////////////////////////////
def main(subscribed_image):
# Argument parsing #################################################################
args = get_args()
cap_device = args.device
cap_width = args.width
cap_height = args.height
use_static_image_mode = args.use_static_image_mode
min_detection_confidence = args.min_detection_confidence
min_tracking_confidence = args.min_tracking_confidence
use_brect = True
# Model load #############################################################
mp_hands = mp.solutions.hands
hands = mp_hands.Hands(
static_image_mode=use_static_image_mode,
max_num_hands=1,
min_detection_confidence=min_detection_confidence,
min_tracking_confidence=min_tracking_confidence,
)
keypoint_classifier = KeyPointClassifier()
point_history_classifier = PointHistoryClassifier()
# Read labels ###########################################################
with open('model/keypoint_classifier/keypoint_classifier_label.csv',
encoding='utf-8-sig') as f:
keypoint_classifier_labels = csv.reader(f)
keypoint_classifier_labels = [
row[0] for row in keypoint_classifier_labels
]
with open(
'model/point_history_classifier/point_history_classifier_label.csv',
encoding='utf-8-sig') as f:
point_history_classifier_labels = csv.reader(f)
point_history_classifier_labels = [
row[0] for row in point_history_classifier_labels
]
# FPS Measurement ########################################################
cvFpsCalc = CvFpsCalc(buffer_len=10)
# Coordinate history #################################################################
history_length = 16
point_history = deque(maxlen=history_length)
# Finger gesture history ################################################
finger_gesture_history = deque(maxlen=history_length)
# ########################################################################
mode = 0
# Camera preparation ###############################################################
#cap = subscribed_image
cap = subscribed_image
# cap = cv.VideoCapture()
while True:
fps = cvFpsCalc.get()
# Process Key (ESC: end) #################################################
key = cv.waitKey(10)
if key == 27: # ESC
break
number, mode = select_mode(key, mode)
# Camera capture #####################################################
image = cap
if image is None:
break
image = cv.flip(image, 1) # Mirror display
debug_image = copy.deepcopy(image)
# Detection implementation #############################################################
image = cv.cvtColor(image, cv.COLOR_BGR2RGB)
image.flags.writeable = False
results = hands.process(image)
image.flags.writeable = True
# ####################################################################
if results.multi_hand_landmarks is not None:
for hand_landmarks, handedness in zip(results.multi_hand_landmarks,
results.multi_handedness):
# Bounding box calculation
brect = calc_bounding_rect(debug_image, hand_landmarks)
# Landmark calculation
landmark_list = calc_landmark_list(debug_image, hand_landmarks)
# Conversion to relative coordinates / normalized coordinates
pre_processed_landmark_list = pre_process_landmark(
landmark_list)
pre_processed_point_history_list = pre_process_point_history(
debug_image, point_history)
# Write to the dataset file
logging_csv(number, mode, pre_processed_landmark_list,
pre_processed_point_history_list)
# Hand sign classification
hand_sign_id = keypoint_classifier(pre_processed_landmark_list)
if hand_sign_id == 2: # Point gesture
point_history.append(landmark_list[8])
else:
point_history.append([0, 0])
# Finger gesture classification
finger_gesture_id = 0
point_history_len = len(pre_processed_point_history_list)
if point_history_len == (history_length * 2):
finger_gesture_id = point_history_classifier(
pre_processed_point_history_list)
# Calculates the gesture IDs in the latest detection
finger_gesture_history.append(finger_gesture_id)
most_common_fg_id = Counter(
finger_gesture_history).most_common()
# Drawing part
debug_image = draw_bounding_rect(use_brect, debug_image, brect)
debug_image = draw_landmarks(debug_image, landmark_list)
debug_image = draw_info_text(
debug_image,
brect,
handedness,
keypoint_classifier_labels[hand_sign_id],
point_history_classifier_labels[most_common_fg_id[0][0]],
)
else:
point_history.append([0, 0])
debug_image = draw_point_history(debug_image, point_history)
debug_image = draw_info(debug_image, fps, mode, number)
# Screen reflection #############################################################
cv.imshow('Hand Gesture Recognition', debug_image)
t = 1
blank_image = np.zeros(shape=[512, 512, 3], dtype=np.uint8)
image = blank_image
cv.destroyAllWindows()
Please could anyone advise, or have any ideas to another means of achieving this?