Upload 3 files

with cv fallback

app.py

@@ -7,7 +7,9 @@ from google.cloud import vision
 _api_key = os.environ["API_KEY"]
 _project_id = os.environ["PROJECT_ID"]
 client = vision.ImageAnnotatorClient(client_options={"quota_project_id": _project_id, "api_key": _api_key})
+# client = vision.ImageAnnotatorClient()
 
+AngTol = 10
 import math
 from scipy.spatial import KDTree
 import io
@@ -15,7 +17,9 @@ from time import time
 from PIL import Image, ImageDraw, ImageFilter
 import numpy as np
 import cv2
-
+import sys
+sys.path.insert(0, ".")
+import classical
 from typing import Union
 
 modelPh = r'corners-best.pt'
@@ -161,7 +165,7 @@ def median_point_of_bounding_box(x1, y1, x2, y2, x3, y3, x4, y4):
 
 def to_numeric(text:str):
     try:
-        return float(text)
+        return float(text.replace(",","."))
     except:
         pass
     return None
@@ -189,7 +193,8 @@ def result_as_validvalue(contents:list[dict])->tuple[list[dict], list[str]]:
 
 distance = lambda a,b : np.sqrt(np.square(np.array(a)-np.array(b)).sum())
 
-def determine_ocr_neighbors(center, valid:list[dict])->tuple[ list, float ]:
+def determine_ocr_neighbors(keypoints, valid:list[dict], nearestIx)->tuple[ list, float ]:
+    center = np.array(keypoints["center"])
     def cosangle(a,b):
         na = np.linalg.norm(a)
         nb = np.linalg.norm(b)
@@ -199,6 +204,8 @@ def determine_ocr_neighbors(center, valid:list[dict])->tuple[ list, float ]:
     # compute angles between values
     values = [valid[0]]
     values[0]["dang"] = 0
+
+    values[0]["ds"] = distance(center, values[0]["mid"])
     rates = []
     angS = 0
     for v in valid[1:]:
@@ -207,26 +214,44 @@ def determine_ocr_neighbors(center, valid:list[dict])->tuple[ list, float ]:
         a = np.array(values[-1]["mid"]) - center 
         b = np.array(v["mid"]) - center
         ang,_ = cosangle(a,b)
-        # if _ <0:
-        #     Warning(f"skipping {u['value']} rot:{_}")
-        #     continue
+        u["rot"] = _
         angS += ang
         u["dang"] = ang
+        # u["ddir"] = rot # counter clockwise?
         u["dvda"] = u["dv"] / ang
         rates.append(u["dvda"])
+        #
+        # u["ds"] = distance(values[-1]["mid"], u["mid"])
+        u["ds"] = distance(center, u["mid"])
         values.append(u)
 
+    if nearestIx[0]==0:
+        rates.insert(0, rates[0])
+
     rates = np.array(rates)
+    # filter outlier rate
+    # ix = np.bitwise_and(rates> np.quantile(rates, 0.05) , rates<np.quantile(rates, 0.95))
+    # rate = rates[ix].mean()
     meanAng = angS/len(valid)
     if len(rates)>=6:
         ix = np.bitwise_and(rates> np.quantile(rates, 0.05) , rates<np.quantile(rates, 0.95))
         if not np.all(~ix):
             rates = rates[ix]
-    rate = rates.mean()
+        rate = rates.mean()
+    elif len(nearestIx)==2:
+        n = [nearestIx[0], nearestIx[1]]
+        rank = np.hstack([np.arange(0,n[0]+1)[::-1], np.arange(n[1],len(rates))-n[1]]).astype(float)
+        weights = np.exp(-2*rank)
+        weights /= weights.sum()
+        rate = np.average(rates, weights=weights)
+    elif len(nearestIx)==1:
+        rate = rates[nearestIx[0]]
+
     rate, meanAng
     return values, rate
 
 
+
 def vec_angle(v1, v2)->tuple[float, bool]:
     vector1 = v1/np.linalg.norm(v1)
     vector2 = v2/np.linalg.norm(v2)
@@ -236,38 +261,83 @@ def vec_angle(v1, v2)->tuple[float, bool]:
 
 def angles_from_tip(keypoints, values, nearestIx):
     center = keypoints["center"]
-    tip = keypoints["tip"] - center
-    v = values[nearestIx[0]]
-    a = v["mid"] - center
-    ang = vec_angle(a,tip)
-    cumsum = 0
-    for i in range(nearestIx[0],-1,-1):
-        values[i]["before"] = abs(ang)+cumsum
-        cumsum += values[i]["dang"]
-
-    v = values[nearestIx[1]]
-    a = v["mid"] - center
-    ang = vec_angle(a,tip)
-    values[nearestIx[1]]["dang"] = 0
-    cumsum = 0
-    for i in range(nearestIx[1], len(values)):
-        cumsum -= values[i]["dang"]
-        values[i]["before"] = -abs(ang)+cumsum
+    tip = keypoints["tip"] - center    
+    N = len(nearestIx)
+    start = nearestIx[0] 
+    if N==2 or (N==1 and nearestIx[0]==len(values)-1):
+        v = values[start]
+        a = v["mid"] - center
+        ang = vec_angle(a,tip)
+        cumsum = 0
+        for i in range(start,-1,-1):
+            values[i]["before"] = abs(ang)+cumsum
+            cumsum += values[i]["dang"]
+
+    if N==2 or (N==1 and nearestIx[0]==0):
+        if N==1:
+            start = nearestIx[0]
+        else:
+            start = nearestIx[1]
+
+        v = values[start]
+        a = v["mid"] - center
+        ang = vec_angle(a,tip)
+
+        values[start]["dang"] = 0
+        cumsum = 0
+        for i in range(start, len(values)):
+            cumsum -= values[i]["dang"]
+            values[i]["before"] = -abs(ang)+cumsum
 
     return values
 
 
+
 def sort_clockwise_with_start(coordinates, x_center, y_center, starting_index):
     angles = [math.atan2(y - y_center, x - x_center) for x, y in coordinates]
     sorted_indices = sorted(range(len(angles)), key=lambda i: (angles[i] - angles[starting_index] + 2 * math.pi) % (2 * math.pi))
     return sorted_indices, angles
 
 def remove_nonrange_value(valid):
-    meanArea = np.mean([e["apchar"] for e in valid])
+    # meanArea = np.mean([e["apchar"] for e in valid])
+    meanArea = np.mean([e["apchar"] for e in valid if "apchar" in e])
     cutoff = 0.5
-    valid = list(filter(lambda e: abs(e["apchar"]-meanArea)/meanArea < cutoff, valid))
+    # valid = list(filter(lambda e: abs(e["apchar"]-meanArea)/meanArea < cutoff, valid))
+    valid = list(filter(lambda e:  True if e["text"]=="tip" else abs(e["apchar"]-meanArea)/meanArea < cutoff, valid))
     return valid
 
+
+
+def check_tip(img, keypoints):
+
+    lines = classical.get_needle_line(np.array(img))
+    if lines is None or len(lines)==0:
+        return False
+    # lines = lines.squeeze()
+    if lines.ndim==1:
+        lines = np.expand_dims(lines,axis=0)
+    # nearest line to center,
+    dist2 = lambda a,b: (a[0]-b[0])**2 + (a[1]-b[1])**2
+    center = keypoints["center"]
+    ds = [ min(dist2(center, e[:2]), dist2(center, e[2:])) for e in lines] # closest line to center
+    ix= np.argsort(ds)
+    ix, ds
+    l = lines[ix][0]
+    a = np.array([l[0]-l[2], l[1]-l[3]])
+    a
+    tip = keypoints["tip"] - center
+    ang = vec_angle(a, tip)
+    if abs(ang) > AngTol:
+        # furthest point from center is tip
+        if dist2(l[:2],center) > dist2(l[2:],center):
+            keypoints["tip"] = l[:2]
+        else:
+            keypoints["tip"] = l[2:]
+        print("new point ", keypoints["tip"])
+        return True
+    return False
+
+
 def get_needle_value(img, keypoints):
 
     tic2 = time()
@@ -282,22 +352,35 @@ def get_needle_value(img, keypoints):
     valid.append({"text":"tip", "mid":keypoints["tip"]})
     ix,an = sort_clockwise_with_start([e["mid"] for e in valid],*keypoints["center"], 0)
     valid = [valid[i] for i in ix]
-    assert valid[-1]["text"]!="tip" and valid[0]["text"]!="tip", "failed to properly detect tip"
+    # assert valid[-1]["text"]!="tip" and valid[0]["text"]!="tip", "failed to properly detect tip"
+    valid = remove_nonrange_value(valid)
+
+    i=0
     nearestIx=[]
     for i,v in enumerate(valid):
         if "tip"==v["text"]:
             nearestIx = [i-1,i]
             valid.pop(i)
             break
+    if len(valid)==nearestIx[1] or -1==nearestIx[0]:
+        # nearestIx[1] = 0 # tip is out of bounds
+        tip = keypoints["tip"] - keypoints["center"]
+        b = valid[0]["mid"] - keypoints["center"]
+        a = valid[-1]["mid"] - keypoints["center"]
+        if abs(vec_angle(tip,a)) < abs(vec_angle(tip, b)):
+            nearestIx = [len(valid)-1]
+        else:
+            nearestIx = [0]
+    # nearest to
     nearestIx = np.array(nearestIx)
-    valid = remove_nonrange_value(valid)
 
     center = np.array(keypoints["center"])
-    values, rate = determine_ocr_neighbors(center, valid)
+    values, rate = determine_ocr_neighbors(keypoints, valid, nearestIx)
     assert len(values)>=2, "failed to find at least 2 OCR values"
 
     # import pandas as pd
     # print(pd.DataFrame.from_dict(values))
+    # print(nearestIx)
 
     # tree = KDTree([v["mid"] for v in values])
     # # find bounding ocr values of tip
@@ -329,7 +412,7 @@ def get_needle_value(img, keypoints):
     # print(f"total took: {toc-tic:.1g}")
     tipValues = np.array(tipValues)
 
-    debug(img, contents, keypoints)
+    # debug(img, contents, keypoints)
 
     startValue= float(values[0]["value"])
     tipvalue= round(float(tipValues[nearestIx].mean()),2)
@@ -365,30 +448,34 @@ def debug(img, contents, keypoints):
 
 
 def predict(img, detect_gauge_first):
+    KPs  = []
     if detect_gauge_first:
         model0 = get_load_PhModel()
         results = model0.predict(img)
-        phimgs,_ = get_corners(results, img)
+        phimgs,KPs = get_corners(results, img)
         if len(phimgs)==0:
             raise gr.Error("no gauge found")
     else:
         phimgs = [img.copy()]
 
     payloads = []
-    for phimg in phimgs:
+    for i,phimg in enumerate(phimgs):
         model = get_load_KpModel()
         phimg = preprocessImg(phimg)
         results = model.predict(phimg)
         keypoints = get_keypoints(results)
-
         angle2tip, totalAngle = calculate_sweep_angles(keypoints)
-
+        angReplaced = check_tip(phimg, keypoints)
         phimg = phimg.filter(ImageFilter.UnsharpMask(radius=3))
         payload = get_needle_value(phimg, keypoints)
         payload["angleToTip"] = round(float(angle2tip),2)
+        if angReplaced:
+            payload["angleToTip"] = None
         payload["totalAngle"] = round(float(totalAngle),2)
         for k,v in payload.items():
             print(k, type(v),v)
+        if len(KPs)>i:
+            payload["bbox"] = {k:v.astype(int).tolist() for k,v in KPs[i].items()}
         payloads.append(payload)
 
     return payloads

classical.py

@@ -0,0 +1,159 @@
+import cv2
+import numpy as np
+linelen = lambda l: np.sqrt((l[0]-l[2])**2 + (l[1]-l[3])**2)
+import math
+
+class HoughBundler:     
+    def __init__(self,min_distance=5,min_angle=2):
+        self.min_distance = min_distance
+        self.min_angle = min_angle
+    
+    def get_orientation(self, line):
+        orientation = math.atan2(abs((line[3] - line[1])), abs((line[2] - line[0])))
+        return math.degrees(orientation)
+
+    def check_is_line_different(self, line_1, groups, min_distance_to_merge, min_angle_to_merge):
+        for group in groups:
+            for line_2 in group:
+                if self.get_distance(line_2, line_1) < min_distance_to_merge:
+                    orientation_1 = self.get_orientation(line_1)
+                    orientation_2 = self.get_orientation(line_2)
+                    if abs(orientation_1 - orientation_2) < min_angle_to_merge:
+                        group.append(line_1)
+                        return False
+        return True
+
+    def distance_point_to_line(self, point, line):
+        px, py = point
+        x1, y1, x2, y2 = line
+
+        def line_magnitude(x1, y1, x2, y2):
+            line_magnitude = math.sqrt(math.pow((x2 - x1), 2) + math.pow((y2 - y1), 2))
+            return line_magnitude
+
+        lmag = line_magnitude(x1, y1, x2, y2)
+        if lmag < 0.00000001:
+            distance_point_to_line = 9999
+            return distance_point_to_line
+
+        u1 = (((px - x1) * (x2 - x1)) + ((py - y1) * (y2 - y1)))
+        u = u1 / (lmag * lmag)
+
+        if (u < 0.00001) or (u > 1):
+            #// closest point does not fall within the line segment, take the shorter distance
+            #// to an endpoint
+            ix = line_magnitude(px, py, x1, y1)
+            iy = line_magnitude(px, py, x2, y2)
+            if ix > iy:
+                distance_point_to_line = iy
+            else:
+                distance_point_to_line = ix
+        else:
+            # Intersecting point is on the line, use the formula
+            ix = x1 + u * (x2 - x1)
+            iy = y1 + u * (y2 - y1)
+            distance_point_to_line = line_magnitude(px, py, ix, iy)
+
+        return distance_point_to_line
+
+    def get_distance(self, a_line, b_line):
+        dist1 = self.distance_point_to_line(a_line[:2], b_line)
+        dist2 = self.distance_point_to_line(a_line[2:], b_line)
+        dist3 = self.distance_point_to_line(b_line[:2], a_line)
+        dist4 = self.distance_point_to_line(b_line[2:], a_line)
+
+        return min(dist1, dist2, dist3, dist4)
+
+    def merge_lines_into_groups(self, lines):
+        groups = []  # all lines groups are here
+        # first line will create new group every time
+        groups.append([lines[0]])
+        # if line is different from existing gropus, create a new group
+        for line_new in lines[1:]:
+            if self.check_is_line_different(line_new, groups, self.min_distance, self.min_angle):
+                groups.append([line_new])
+
+        return groups
+
+    def merge_line_segments(self, lines):
+        orientation = self.get_orientation(lines[0])
+      
+        if(len(lines) == 1):
+            return np.block([[lines[0][:2], lines[0][2:]]])
+
+        points = []
+        for line in lines:
+            points.append(line[:2])
+            points.append(line[2:])
+        if 45 < orientation <= 90:
+            #sort by y
+            points = sorted(points, key=lambda point: point[1])
+        else:
+            #sort by x
+            points = sorted(points, key=lambda point: point[0])
+
+        p0 = np.array(points[:2]).mean(axis=0)
+        p1 = np.array(points[-2:]).mean(axis=0)
+        return np.block([[p0,p1]]).astype(int)
+        # return np.block([[points[0],points[-1]]])
+
+    def process_lines(self, lines):
+        lines_horizontal  = []
+        lines_vertical  = []
+  
+        for line_i in [l[0] for l in lines]:
+            orientation = self.get_orientation(line_i)
+            # if vertical
+            if 45 < orientation <= 90:
+                lines_vertical.append(line_i)
+            else:
+                lines_horizontal.append(line_i)
+
+        lines_vertical  = sorted(lines_vertical , key=lambda line: line[1])
+        lines_horizontal  = sorted(lines_horizontal , key=lambda line: line[0])
+        merged_lines_all = []
+
+        # for each cluster in vertical and horizantal lines leave only one line
+        for i in [lines_horizontal, lines_vertical]:
+            if len(i) > 0:
+                groups = self.merge_lines_into_groups(i)
+                merged_lines = []
+                for group in groups:
+                    merged_lines.append(self.merge_line_segments(group))
+                merged_lines_all.extend(merged_lines)
+                    
+        return np.asarray(merged_lines_all)
+
+
+def get_needle_line(im)->list:
+    # fn = "./0.png"
+    # # load grayscale image
+    # im = cv2.imread(fn)
+    gray_im = cv2.cvtColor(im, cv2.COLOR_RGB2GRAY)
+    sz = 640
+    h,w,_ = im.shape
+    _hf = h/sz
+    _wf = w/sz
+    gray_im = cv2.resize(gray_im, (sz,sz))
+    
+    blur = cv2.GaussianBlur(gray_im, (0,0), 5)
+    
+    # edges = cv2.Canny(blur, 50, 100)
+    edges = cv2.Canny(blur, 20, 60)
+    
+    rectKernel = cv2.getStructuringElement(cv2.MORPH_RECT, (11, 19))
+    hat = cv2.morphologyEx(edges, cv2.MORPH_BLACKHAT, rectKernel)
+    # hat = cv2.morphologyEx(edges, cv2.MORPH_TOPHAT, rectKernel)
+    # k = cv2.getStructuringElement(cv2.MORPH_ERODE, (13,13))
+    k = None
+    hat = cv2.erode(hat,k, iterations=2)
+
+    minLineLength = 60
+    maxLineGap = 10
+    plines = cv2.HoughLinesP(image=edges, rho=3, theta=np.pi / 180, threshold=10,minLineLength=minLineLength, maxLineGap=maxLineGap)  # rho is set to 3 to detect more lines, easier to get more then filter them out later
+    if len(plines)<=1:
+        return plines.squeeze() * [_wf, _hf, _wf, _hf]
+    
+    bundler = HoughBundler(min_distance=120,min_angle=5)
+    clines = bundler.process_lines(plines)
+    return clines.squeeze() * [_wf, _hf, _wf, _hf]

keypoints-best.pt

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:d583485a30cd58e986231e7a02b84ce86e117d7eb48d4b5a901e4bada55319ac
-size 6408962
+oid sha256:8b9b6ac6b3e5dd73a4e00af18365b13e0607cb8e4b00cfd9189e005b86103124
+size 6409410