Major refactoring for better usability
1. Created interface classes for reading video streams (in imagereaders.py) 2. Created a class for ball detection (for reusability) 3. Reworked colorpicker, now it is possible to choose the mode of operation from command line (available are live from Nao, from video file or from webcam). It is possible to capture only the first image of a stream and work on it. Colorpicker now can save the settings on exit or load settings on startup.
This commit is contained in:
@@ -1,137 +1,113 @@
|
||||
from __future__ import print_function
|
||||
from __future__ import division
|
||||
|
||||
import json
|
||||
import cv2
|
||||
import numpy as np
|
||||
import imutils
|
||||
from naoqi import ALProxy
|
||||
from imagereaders import NaoImageReader, VideoReader
|
||||
from collections import deque
|
||||
|
||||
|
||||
# Nao configuration
|
||||
nao_ip = '192.168.0.11'
|
||||
nao_port = 9559
|
||||
res = (1, (240, 320)) # NAOQi code and acutal resolution
|
||||
fps = 30
|
||||
cam_id = 1 # 0 := top, 1 := bottom
|
||||
|
||||
# Recognition stuff
|
||||
red_lower = (0, 185, 170) # HSV coded red interval
|
||||
red_upper = (2, 255, 255)
|
||||
min_radius = 5
|
||||
resized_width = None # Maybe we need it maybe don't (None if don't)
|
||||
|
||||
def get_frame_nao(cam_proxy, subscriber, width, height):
|
||||
result = cam_proxy.getImageRemote(subscriber)
|
||||
cam_proxy.releaseImage(subscriber)
|
||||
if result == None:
|
||||
raise RuntimeError('cannot capture')
|
||||
elif result[6] == None:
|
||||
raise ValueError('no image data string')
|
||||
else:
|
||||
return np.frombuffer(result[6], dtype=np.uint8).reshape(
|
||||
height, width, 3
|
||||
)
|
||||
# i = 0
|
||||
# for y in range(res[1][0]):
|
||||
# for x in range(res[1][1]): # columnwise
|
||||
# image.itemset((y, x, 0), values[i + 0])
|
||||
# image.itemset((y, x, 1), values[i + 1])
|
||||
# image.itemset((y, x, 2), values[i + 2])
|
||||
# i += 3
|
||||
# return image
|
||||
|
||||
|
||||
def find_colored_ball(frame, hsv_lower, hsv_upper, min_radius):
|
||||
hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
|
||||
class BallFinder(object):
|
||||
|
||||
# construct a mask for the color "green", then perform a series of
|
||||
# dilations and erosions to remove any small blobs left in the mask
|
||||
mask = cv2.inRange(hsv, hsv_lower, hsv_upper)
|
||||
mask = cv2.erode(mask, None, iterations=2)
|
||||
mask = cv2.dilate(mask, None, iterations=2)
|
||||
cv2.imshow('ball_mask', mask)
|
||||
cv2.waitKey(1)
|
||||
def __init__(self, hsv_lower, hsv_upper, min_radius, width):
|
||||
|
||||
# find contours in the mask and initialize the current
|
||||
# (x, y) center of the ball
|
||||
cnts = cv2.findContours(mask.copy(), cv2.RETR_EXTERNAL,
|
||||
cv2.CHAIN_APPROX_SIMPLE)[-2]
|
||||
self.hsv_lower = hsv_lower
|
||||
self.hsv_upper = hsv_upper
|
||||
self.min_radius = min_radius
|
||||
self.width = width
|
||||
self.history = deque(maxlen=64)
|
||||
self.last_center = None
|
||||
self.last_radius = None
|
||||
|
||||
# only proceed if at least one contour was found
|
||||
if len(cnts) == 0:
|
||||
return None
|
||||
cv2.namedWindow('ball_mask')
|
||||
cv2.namedWindow('Frame')
|
||||
|
||||
# find the largest contour in the mask, then use it to compute
|
||||
# the minimum enclosing circle and centroid
|
||||
c = max(cnts, key=cv2.contourArea)
|
||||
((x, y), radius) = cv2.minEnclosingCircle(c)
|
||||
def find_colored_ball(self, frame):
|
||||
hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
|
||||
|
||||
if radius < min_radius:
|
||||
return None
|
||||
# construct a mask for the color, then perform a series of
|
||||
# dilations and erosions to remove any small blobs left in the mask
|
||||
mask = cv2.inRange(hsv, self.hsv_lower, self.hsv_upper)
|
||||
mask = cv2.erode(mask, None, iterations=2)
|
||||
mask = cv2.dilate(mask, None, iterations=2)
|
||||
cv2.imshow('ball_mask', mask)
|
||||
|
||||
M = cv2.moments(c)
|
||||
center = (int(M["m10"] / M["m00"]),int(M["m01"] // M["m00"]))
|
||||
return center, int(radius)
|
||||
# find contours in the mask and initialize the current
|
||||
# (x, y) center of the ball
|
||||
cnts = cv2.findContours(mask.copy(), cv2.RETR_EXTERNAL,
|
||||
cv2.CHAIN_APPROX_SIMPLE)[-2]
|
||||
|
||||
# only proceed if at least one contour was found
|
||||
if len(cnts) == 0:
|
||||
return None
|
||||
|
||||
def draw_ball_markers(frame, center, radius, history):
|
||||
# draw the enclosing circle and ball's centroid on the frame,
|
||||
if center is not None and radius is not None:
|
||||
cv2.circle(frame, center, radius, (255, 255, 0), 1)
|
||||
cv2.circle(frame, center, 5, (0, 255, 0), -1)
|
||||
# find the largest contour in the mask, then use it to compute
|
||||
# the minimum enclosing circle and centroid
|
||||
c = max(cnts, key=cv2.contourArea)
|
||||
((x, y), radius) = cv2.minEnclosingCircle(c)
|
||||
|
||||
# loop over the set of tracked points
|
||||
for i in range(1, len(history)):
|
||||
# if either of the tracked points are None, ignore them
|
||||
if history[i - 1] is None or history[i] is None:
|
||||
continue
|
||||
# otherwise, compute the thickness of the line and
|
||||
# draw the connecting lines
|
||||
thickness = int(np.sqrt(64 / float(i + 1)) * 2.5)
|
||||
cv2.line(frame, history[i - 1], history[i], (0, 255, 0), thickness)
|
||||
if radius < self.min_radius:
|
||||
return None
|
||||
|
||||
return frame
|
||||
M = cv2.moments(c)
|
||||
center = (int(M["m10"] / M["m00"]),int(M["m01"] // M["m00"]))
|
||||
return center, int(radius)
|
||||
|
||||
def next_frame(self, frame):
|
||||
# maybe resize the frame, maybe blur it
|
||||
if self.width is not None:
|
||||
frame = imutils.resize(frame, width=self.width)
|
||||
try:
|
||||
self.last_center, self.last_radius = self.find_colored_ball(frame)
|
||||
except TypeError: # No red ball found and function returned None
|
||||
self.last_center, self.last_radius = None, None
|
||||
|
||||
def nao_demo():
|
||||
cv2.namedWindow('ball_mask')
|
||||
cv2.namedWindow('Frame')
|
||||
self.history.appendleft(self.last_center)
|
||||
self.draw_ball_markers(frame)
|
||||
|
||||
vd_proxy = ALProxy('ALVideoDevice', nao_ip, nao_port)
|
||||
cam_subscriber = vd_proxy.subscribeCamera(
|
||||
"ball_finder", cam_id, res[0], 13, fps
|
||||
)
|
||||
history = deque(maxlen=64)
|
||||
# show the frame to screen
|
||||
cv2.imshow("Frame", frame)
|
||||
return cv2.waitKey(2)
|
||||
|
||||
try:
|
||||
while True:
|
||||
frame = get_frame_nao(vd_proxy, cam_subscriber, res[1][1],
|
||||
res[1][0])
|
||||
def draw_ball_markers(self, frame):
|
||||
# draw the enclosing circle and ball's centroid on the frame,
|
||||
if self.last_center is not None and self.last_radius is not None:
|
||||
cv2.circle(frame, self.last_center, self.last_radius,
|
||||
(255, 255, 0), 1)
|
||||
cv2.circle(frame, self.last_center, 5, (0, 255, 0), -1)
|
||||
|
||||
# maybe resize the frame, maybe blur it
|
||||
if resized_width is not None:
|
||||
frame = imutils.resize(frame, width=resized_width)
|
||||
# blurred = cv2.GaussianBlur(frame, (11, 11), 0)
|
||||
# loop over the set of tracked points
|
||||
for i in range(1, len(self.history)):
|
||||
# if either of the tracked points are None, ignore them
|
||||
if self.history[i - 1] is None or self.history[i] is None:
|
||||
continue
|
||||
# otherwise, compute the thickness of the line and
|
||||
# draw the connecting lines
|
||||
thickness = int(np.sqrt(64 / float(i + 1)) * 2.5)
|
||||
cv2.line(frame, self.history[i - 1], self.history[i],
|
||||
(0, 255, 0), thickness)
|
||||
|
||||
try:
|
||||
center, radius = find_colored_ball(
|
||||
frame, red_lower, red_upper, min_radius
|
||||
)
|
||||
history.appendleft(center)
|
||||
draw_ball_markers(frame, center, radius, history)
|
||||
except TypeError: # No red ball found and function returned None
|
||||
history.appendleft(None)
|
||||
draw_ball_markers(frame, None, None, history)
|
||||
return frame
|
||||
|
||||
# show the frame to screen
|
||||
cv2.imshow("Frame", frame)
|
||||
cv2.waitKey(1)
|
||||
|
||||
finally:
|
||||
vd_proxy.unsubscribe(cam_subscriber)
|
||||
cv2.destroyAllWindows()
|
||||
def load_hsv_config(self, filename):
|
||||
with open(filename) as f:
|
||||
hsv = json.load(f)
|
||||
self.hsv_lower = tuple(map(hsv.get, ('low_h', 'low_s', 'low_v')))
|
||||
self.hsv_upper = tuple(map(hsv.get, ('high_h', 'high_s', 'high_v')))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
nao_demo()
|
||||
# video = NaoImageReader('192.168.0.11')
|
||||
video = VideoReader(0, loop=True)
|
||||
finder = BallFinder(red_lower, red_upper, 5, None)
|
||||
try:
|
||||
while True:
|
||||
finder.next_frame(video.get_frame())
|
||||
finally:
|
||||
video.close()
|
||||
|
||||
Reference in New Issue
Block a user