Read water without sending
This commit is contained in:
Simon Milvert
parent
7aa69e63ac
commit
738161b0da
|
|
@ -1,239 +0,0 @@
|
|||
import math
|
||||
import sys
|
||||
import cv2
|
||||
import os
|
||||
import numpy as np
|
||||
|
||||
|
||||
class WaterMeter(object):
|
||||
|
||||
def __init__(self, url, img=None, debug=False):
|
||||
self.url = url
|
||||
self.img = img
|
||||
self.debug = debug
|
||||
|
||||
def read(self):
|
||||
if self.img:
|
||||
img = cv2.imread(self.img)
|
||||
return img
|
||||
|
||||
def loop(self):
|
||||
img = self.read()
|
||||
needle = self.find_red_needle(img)
|
||||
#needle = self.get_contours(needle, img)
|
||||
dials = self.find_circle(img)
|
||||
|
||||
for dial in dials:
|
||||
ulx = int(dial[0])
|
||||
uly = int(dial[1])
|
||||
radius = int(dial[2]) * 3
|
||||
cut = img[uly - radius: uly + radius,
|
||||
ulx - radius: ulx + radius]
|
||||
needle_cut = needle[uly - radius: uly + radius,
|
||||
ulx - radius: ulx + radius]
|
||||
self.findAngle(cut, needle_cut, (radius, radius), radius * 2)
|
||||
if self.debug:
|
||||
cv2.imshow('image', img)
|
||||
cv2.waitKey(0)
|
||||
|
||||
|
||||
@staticmethod
|
||||
def line(p1, p2):
|
||||
A = (p1[1] - p2[1])
|
||||
B = (p2[0] - p1[0])
|
||||
C = (p1[0] * p2[1] - p2[0] * p1[1])
|
||||
return A, B, -C
|
||||
|
||||
@staticmethod
|
||||
def intersection(L1, L2):
|
||||
D = L1[0] * L2[1] - L1[1] * L2[0]
|
||||
Dx = L1[2] * L2[1] - L1[1] * L2[2]
|
||||
Dy = L1[0] * L2[2] - L1[2] * L2[0]
|
||||
if D != 0:
|
||||
x = Dx / D
|
||||
y = Dy / D
|
||||
return x, y
|
||||
else:
|
||||
return False
|
||||
|
||||
def find_circle(self, img):
|
||||
img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
||||
gradient = cv2.HOUGH_GRADIENT
|
||||
# Find suitable number by running detect_circle.py
|
||||
circles = cv2.HoughCircles(img, gradient, 1.2, 40,
|
||||
param1=50, param2=50, minRadius=50,
|
||||
maxRadius=100)
|
||||
if circles is not None:
|
||||
circles = np.round(circles[0, :]).astype("int")
|
||||
dials = []
|
||||
for (x, y, r) in circles:
|
||||
dials.append([x, y, r])
|
||||
if self.debug:
|
||||
# draw the circle
|
||||
cv2.circle(img, (x, y), r, (0, 255, 0), 4)
|
||||
# draw a rectangle in the middle
|
||||
cv2.rectangle(img, (x - 5, y - 5), (x + 5, y + 5),
|
||||
(0, 128, 255),
|
||||
-1)
|
||||
if self.debug:
|
||||
cv2.imshow('circle', img)
|
||||
cv2.waitKey(0)
|
||||
|
||||
return dials
|
||||
|
||||
def find_red_needle(self, img):
|
||||
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
|
||||
# Find suitable hsv number by running detect_hsv.py
|
||||
lower_red_hue = self.create_hue_mask(hsv, [0, 100, 100], [10, 255, 255])
|
||||
higher_red_hue = self.create_hue_mask(hsv, [170, 100, 100], [179, 255, 255])
|
||||
|
||||
mask = cv2.bitwise_or(lower_red_hue, higher_red_hue)
|
||||
needle = cv2.GaussianBlur(mask, (5, 5), 0)
|
||||
return needle
|
||||
|
||||
def get_contours(self, img, org):
|
||||
ret, thresh = cv2.threshold(img, 127, 255, 0)
|
||||
contours, hierarchy = cv2.findContours(thresh, cv2.RETR_TREE,
|
||||
cv2.CHAIN_APPROX_SIMPLE)
|
||||
|
||||
if self.debug:
|
||||
for contour in contours:
|
||||
peri = cv2.arcLength(contour, True)
|
||||
approx = cv2.approxPolyDP(contour, 0.04 * peri, True)
|
||||
|
||||
cv2.drawContours(org, [approx], -1, (0, 0, 255), 3)
|
||||
|
||||
#cv2.drawContours(org, contours, -1, (0, 255, 0), 3)
|
||||
cv2.imshow('needle', org)
|
||||
cv2.waitKey(0)
|
||||
|
||||
return contours
|
||||
|
||||
@staticmethod
|
||||
def draw_line(lines, img):
|
||||
for r, theta in lines[0]:
|
||||
# Stores the value of cos(theta) in a
|
||||
a = np.cos(theta)
|
||||
|
||||
# Stores the value of sin(theta) in b
|
||||
b = np.sin(theta)
|
||||
|
||||
# x0 stores the value rcos(theta)
|
||||
x0 = a * r
|
||||
|
||||
# y0 stores the value rsin(theta)
|
||||
y0 = b * r
|
||||
|
||||
# x1 stores the rounded off value of (rcos(theta)-1000sin(theta))
|
||||
x1 = int(x0 + 1000 * (-b))
|
||||
|
||||
# y1 stores the rounded off value of (rsin(theta)+1000cos(theta))
|
||||
y1 = int(y0 + 1000 * (a))
|
||||
|
||||
# x2 stores the rounded off value of (rcos(theta)+1000sin(theta))
|
||||
x2 = int(x0 - 1000 * (-b))
|
||||
|
||||
# y2 stores the rounded off value of (rsin(theta)-1000cos(theta))
|
||||
y2 = int(y0 - 1000 * (a))
|
||||
|
||||
# cv2.line draws a line in img from the point(x1,y1) to (x2,y2).
|
||||
# (0,0,255) denotes the colour of the line to be
|
||||
# drawn. In this case, it is red.
|
||||
cv2.line(img, (x1, y1), (x2, y2), (0, 0, 255), 2)
|
||||
|
||||
return img
|
||||
|
||||
def findAngle(self, img, redimg, center, width):
|
||||
### lines
|
||||
edges = cv2.Canny(redimg, 100, 200, apertureSize=3)
|
||||
lines = cv2.HoughLinesP(image=edges,
|
||||
rho=1,
|
||||
theta=np.pi / 90,
|
||||
threshold=15,
|
||||
minLineLength=width / 4,
|
||||
maxLineGap=50)
|
||||
|
||||
tip = None
|
||||
maxlen = 0
|
||||
if lines is None:
|
||||
print("No lines found")
|
||||
|
||||
cv2.imshow('error', img)
|
||||
cv2.waitKey()
|
||||
cv2.destroyAllWindows()
|
||||
|
||||
else:
|
||||
pl = None
|
||||
img = self.draw_line(lines, img)
|
||||
cv2.imshow('error', img)
|
||||
cv2.waitKey()
|
||||
cv2.destroyAllWindows()
|
||||
# print "%d lines" % len(lines)
|
||||
for x in range(0, len(lines)):
|
||||
for x1, y1, x2, y2 in lines[x]:
|
||||
l = self.line([x1, y1], [x2, y2]);
|
||||
cv2.line(img, (x1, y1), (x2, y2), (0, 255, 0), 2)
|
||||
# print "line: %d,%d %d,%d" % (x1, y1, x2, y2)
|
||||
|
||||
# see if it intersects any other line
|
||||
for y in range(0, len(lines)):
|
||||
for xx1, yy1, xx2, yy2 in lines[y]:
|
||||
l2 = self.line([xx1, yy1], [xx2, yy2]);
|
||||
if l2 is l:
|
||||
continue
|
||||
r = self.intersection(l, l2)
|
||||
if r and r[0] > 0 and r[1] > 0:
|
||||
dist = math.sqrt((r[0] - center[0]) ** 2 + (
|
||||
r[1] - center[1]) ** 2)
|
||||
# print "intersection %d,%d at distance %d" % ( r[0], r[1], dist)
|
||||
# cv2.circle(img,(r[0],r[1]),2,(0,0,255),2)
|
||||
if dist > maxlen and dist < width / 2:
|
||||
tip = r
|
||||
maxlen = dist
|
||||
|
||||
|
||||
|
||||
# print "chosen intersection: %d,%d at distance %d" % ( tip[0], tip[1], maxlen)
|
||||
#cv2.line(img, (tip[0], tip[1]), (center[0], center[1]), (255, 0, 255),
|
||||
# 2)
|
||||
|
||||
xlen = tip[0] - center[0]
|
||||
ylen = center[1] - tip[1]
|
||||
rad = math.atan2(ylen, xlen)
|
||||
deg = math.degrees(rad)
|
||||
# print "angle deg:", deg
|
||||
# print "angle rad:", rad
|
||||
|
||||
if deg < 0:
|
||||
percent = (90 + abs(deg)) / 360
|
||||
elif deg < 90:
|
||||
percent = (90 - deg) / 360
|
||||
else:
|
||||
percent = (450 - deg) / 360
|
||||
|
||||
# print "percent", math.trunc(percent * 100)
|
||||
string = "%d%%" % math.trunc(percent * 100)
|
||||
cv2.putText(img, string, (center[0] - width / 5, center[1] - width / 3),
|
||||
cv2.FONT_HERSHEY_SIMPLEX, 0.7,
|
||||
(255, 255, 255), 2)
|
||||
|
||||
return math.trunc(percent * 100)
|
||||
|
||||
@staticmethod
|
||||
def create_hue_mask(image, lower_color, upper_color):
|
||||
lower = np.array(lower_color, np.uint8)
|
||||
upper = np.array(upper_color, np.uint8)
|
||||
|
||||
# Create a mask from the colors
|
||||
mask = cv2.inRange(image, lower, upper)
|
||||
return mask
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
water_meter = WaterMeter('', img='capture_1.jpg', debug=True)
|
||||
water_meter.loop()
|
||||
|
||||
|
|
@ -0,0 +1,225 @@
|
|||
import math
|
||||
import sys
|
||||
import time
|
||||
|
||||
import cv2
|
||||
import os
|
||||
import numpy as np
|
||||
import imutils
|
||||
from scipy.spatial import distance as dist
|
||||
|
||||
|
||||
class WaterMeter(object):
|
||||
|
||||
def __init__(self, url, img=None, debug=False):
|
||||
self.url = url
|
||||
self.img = img
|
||||
self.debug = debug
|
||||
self.cap = None
|
||||
self._quit = False
|
||||
|
||||
def _init_camera(self):
|
||||
self.cap = cv2.VideoCapture()
|
||||
self.cap.open(self.url)
|
||||
|
||||
if self.cap is None:
|
||||
print("Can't access camera")
|
||||
return None
|
||||
|
||||
if not self.cap.isOpened():
|
||||
print("Can't open camera")
|
||||
return None
|
||||
|
||||
return self.cap
|
||||
|
||||
def read(self):
|
||||
|
||||
img = cv2.imread(self.img)
|
||||
self.get_degree(img)
|
||||
|
||||
def loop(self):
|
||||
bad_frame = 0
|
||||
while not self._quit:
|
||||
if self._quit:
|
||||
break
|
||||
|
||||
if bad_frame > 100:
|
||||
if self.cap is not None:
|
||||
self.cap.release()
|
||||
self.cap = None
|
||||
bad_frames = 0
|
||||
|
||||
# initializing connection to camera
|
||||
if self.cap is None:
|
||||
self._init_camera()
|
||||
|
||||
ret, current_frame = self.cap.read()
|
||||
# the connection broke, or the stream came to an end
|
||||
if (not ret) or (current_frame is None):
|
||||
# ToDo Logging error frame
|
||||
bad_frame += 1
|
||||
continue
|
||||
else:
|
||||
bad_frame = 0
|
||||
|
||||
degree, img = self.get_degree(current_frame)
|
||||
if degree < 0:
|
||||
percent = (90 + abs(degree)) / 360
|
||||
elif degree < 90 and degree != -1:
|
||||
percent = (90 - degree) / 360
|
||||
else:
|
||||
percent = (450 - degree) / 360
|
||||
print(f"Degree is {degree} and percent is {percent * 100}")
|
||||
cv2.imshow("Degree", current_frame)
|
||||
cv2.waitKey(1)
|
||||
|
||||
def stop(self):
|
||||
self._quit = True
|
||||
|
||||
if self.cap is not None:
|
||||
self.cap.release()
|
||||
|
||||
def get_degree(self, frame):
|
||||
img = frame
|
||||
dials = self.find_circle(img)
|
||||
# Init value out of range
|
||||
deg = -1
|
||||
dial_one_litre = []
|
||||
if not dials:
|
||||
return -1, img
|
||||
for dial in dials:
|
||||
# Using only one circle (1 litre) left bottom
|
||||
if not dial_one_litre:
|
||||
dial_one_litre = dial
|
||||
if dial[0] < dial_one_litre[0]:
|
||||
dial_one_litre = dial
|
||||
|
||||
ulx = int(dial_one_litre[0])
|
||||
uly = int(dial_one_litre[1])
|
||||
radius = int(dial_one_litre[2])
|
||||
# Only using image in dial
|
||||
cut = img[uly - radius: uly + radius, ulx - radius: ulx + radius]
|
||||
|
||||
# Find needle in dial
|
||||
needle = self.find_red_needle(cut)
|
||||
extreme_points = self.get_contours(needle, cut)
|
||||
if not extreme_points:
|
||||
return -1, img
|
||||
deg, point = self.find_angle(extreme_points, cut)
|
||||
cv2.line(img,
|
||||
((ulx - radius) + point['x'], uly - radius + point['y']),
|
||||
(ulx, uly), (0, 255, 255), 2)
|
||||
|
||||
if self.debug:
|
||||
cv2.imshow('image', img)
|
||||
cv2.waitKey(0)
|
||||
return deg, img
|
||||
|
||||
def find_circle(self, img):
|
||||
img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
||||
gradient = cv2.HOUGH_GRADIENT
|
||||
# Find suitable number by running detect_circle.py
|
||||
circles = cv2.HoughCircles(img, gradient, dp=1.26, minDist=42,
|
||||
param1=52, param2=43, minRadius=67,
|
||||
maxRadius=99)
|
||||
if circles is not None:
|
||||
circles = np.round(circles[0, :]).astype("int")
|
||||
dials = []
|
||||
for (x, y, r) in circles:
|
||||
dials.append([x, y, r])
|
||||
if self.debug:
|
||||
# draw the circle
|
||||
cv2.circle(img, (x, y), r, (0, 255, 0), 4)
|
||||
# draw a rectangle in the middle
|
||||
cv2.rectangle(img, (x - 5, y - 5), (x + 5, y + 5),
|
||||
(0, 128, 255),
|
||||
-1)
|
||||
if self.debug:
|
||||
cv2.imshow('circle', img)
|
||||
cv2.waitKey(0)
|
||||
|
||||
return dials
|
||||
return None
|
||||
|
||||
def find_red_needle(self, img):
|
||||
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
|
||||
# Find suitable hsv number by running detect_hsv.py
|
||||
lower_red_hue = self.create_hue_mask(hsv, [0, 100, 100], [10, 255, 255])
|
||||
higher_red_hue = self.create_hue_mask(hsv, [170, 80, 110], [179, 255, 255])
|
||||
|
||||
mask = cv2.bitwise_or(lower_red_hue, higher_red_hue)
|
||||
needle = cv2.GaussianBlur(mask, (5, 5), 0)
|
||||
if self.debug:
|
||||
cv2.imshow('circle', needle)
|
||||
cv2.waitKey(0)
|
||||
return needle
|
||||
|
||||
def get_contours(self, img, org):
|
||||
ret, thresh = cv2.threshold(img, 127, 255, 0)
|
||||
contours = cv2.findContours(thresh, cv2.RETR_TREE,
|
||||
cv2.CHAIN_APPROX_SIMPLE)
|
||||
cnts = imutils.grab_contours(contours)
|
||||
try:
|
||||
c = max(cnts, key=cv2.contourArea)
|
||||
except ValueError:
|
||||
print("Not finding any needle")
|
||||
timestr = time.strftime("%Y%m%d-%H%M%S")
|
||||
cv2.imwrite('error_neddle' + timestr + '.png', img)
|
||||
return None
|
||||
# determine the most extreme points along the contour
|
||||
ext_left = tuple(c[c[:, :, 0].argmin()][0])
|
||||
ext_right = tuple(c[c[:, :, 0].argmax()][0])
|
||||
ext_top = tuple(c[c[:, :, 1].argmin()][0])
|
||||
ext_bot = tuple(c[c[:, :, 1].argmax()][0])
|
||||
|
||||
extreme_points = (ext_left, ext_right, ext_top, ext_bot)
|
||||
|
||||
if self.debug:
|
||||
cv2.drawContours(org, [c], -1, (0, 255, 255), 2)
|
||||
cv2.circle(org, ext_left, 8, (0, 0, 255), -1)
|
||||
cv2.circle(org, ext_right, 8, (0, 255, 0), -1)
|
||||
cv2.circle(org, ext_top, 8, (255, 0, 0), -1)
|
||||
cv2.circle(org, ext_bot, 8, (255, 255, 0), -1)
|
||||
# show the output image
|
||||
cv2.imshow("Image", org)
|
||||
cv2.waitKey(0)
|
||||
|
||||
return extreme_points
|
||||
|
||||
@staticmethod
|
||||
def create_hue_mask(image, lower_color, upper_color):
|
||||
lower = np.array(lower_color, np.uint8)
|
||||
upper = np.array(upper_color, np.uint8)
|
||||
|
||||
# Create a mask from the colors
|
||||
mask = cv2.inRange(image, lower, upper)
|
||||
return mask
|
||||
|
||||
def find_angle(self, extreme_points, cut):
|
||||
length_from_centre = {}
|
||||
height, width, _ = cut.shape
|
||||
|
||||
for (x, y) in extreme_points:
|
||||
D = dist.euclidean((x, y), (int(height/2), int(width/2)))
|
||||
if not length_from_centre:
|
||||
length_from_centre = {'x': x, 'y': y, 'distance': D}
|
||||
elif D > length_from_centre['distance']:
|
||||
length_from_centre = {'x': x, 'y': y, 'distance': D}
|
||||
|
||||
if self.debug:
|
||||
cv2.line(cut, (length_from_centre['x'], length_from_centre['y']),
|
||||
(int(height / 2), int(width / 2)), (0, 0, 255), 2)
|
||||
cv2.imshow('length', cut)
|
||||
cv2.waitKey(0)
|
||||
|
||||
rad = math.atan2(length_from_centre['y'], length_from_centre['x'])
|
||||
deg = math.degrees(rad)
|
||||
|
||||
return deg, length_from_centre
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
water_meter = WaterMeter('http://10.0.0.22:81', img='capture_4.png', debug=False)
|
||||
water_meter.loop()
|
||||
#water_meqter.read()
|
||||
|
||||
Loading…
Reference in New Issue