Upload 8 files

.gitignore

@@ -0,0 +1,2 @@
+gradio_cached_examples/
+temp*.jpg

app.py

@@ -0,0 +1,367 @@
+#
+import gradio as gr
+
+from ultralytics import YOLO
+
+from google.cloud import vision
+
+client = vision.ImageAnnotatorClient()
+
+import math
+from scipy.spatial import KDTree
+import io
+from time import time
+from PIL import Image, ImageDraw
+import numpy as np
+import cv2
+
+from typing import Union
+
+modelPh = r'corners-best.pt'
+
+model1DIM = 640
+keypointModel = r'keypoints-best.pt'
+
+
+_examples = ["example1.jpg", "example2.jpg"]
+
+
+def unwarp_image(warped_image, src_points, dst_points, output_width, output_height):
+    src_pts = np.array(src_points).astype(np.float64)
+    dst_pts = np.array(dst_points).astype(np.float64)
+
+    homography, mask = cv2.findHomography(src_pts, dst_pts)
+    unwarped_image = cv2.warpPerspective(
+        np.array(warped_image), homography, (output_width, output_height)
+    )
+    unwarped_image = Image.fromarray(unwarped_image)
+
+    return unwarped_image
+
+
+model0 = None
+
+def get_load_PhModel():
+    global model0
+    if model0 ==None:
+        tic = time()
+        model0 = YOLO(modelPh)  # load a custom model
+        print(f"model0 load took: {time()-tic:.2g}")
+    return model0
+
+
+def get_corners(results:list, img):
+    global model1DIM
+    #  keypoints ie corners for homography
+    KP = "topLeft topRight bottomRight bottomLeft".split()
+    r = results[0]
+    kpco = r.keypoints.xy.cpu().squeeze()
+    assert len(kpco)>0, "not found"
+    keypoints = {k:v.numpy() for v,k in zip(kpco,KP)}
+    sz = model1DIM
+    dstCorners = np.array([(0,0),(sz,0),(sz,sz),(0,sz)])
+    planar = unwarp_image(img, np.array(list(keypoints.values())),dstCorners, sz,sz)
+    # planar.save("temp-ph.jpg")
+    return planar, keypoints
+
+
+model = None
+
+def get_load_KpModel():
+    global model
+    if model == None:
+        tic = time()
+        model = YOLO(keypointModel)  # load a custom model
+        print(f"model load took: {time()-tic:.2g}")
+    return model
+
+def preprocessImg(planar):
+    img = planar.convert('RGB').copy()
+    w,h = img.size
+    smalldl = abs(w-h)/h <0.05
+    _ = max(w,h)
+    DIM = w
+    if w!=h and smalldl:
+        img = img.resize((_,_))
+    elif w!=h:
+        img = img.resize((_,_))
+
+    return img
+
+
+def get_keypoints(results:list):
+    assert len(results) ==1, "found multiple dials. expected only 1"
+    r = results[0]
+    # ordering
+    kp = "start_kp center end_kp tip".split()
+    kpco = r.keypoints.xy.cpu().squeeze()
+    keypoints = {k:v.numpy() for v,k in zip(kpco,kp)}
+    assert len(keypoints["center"])==2, "center keypoint not found"
+    assert len(keypoints["tip"])==2, "tip keypoint not found"
+    return keypoints
+
+def cosangle(a,b, ignoreRot=False):
+    na = np.linalg.norm(a)
+    nb = np.linalg.norm(b)
+    angle2tip = np.rad2deg(np.arccos(np.dot(a, b)/(na*nb)))
+    angle2tip
+    rotdir = np.cross(a,b) < 0
+    if rotdir and not ignoreRot:
+        return 360-angle2tip
+    return angle2tip
+
+
+def calculate_sweep_angles(keypoints:dict): 
+    # get sweep angles start->tip
+    a = keypoints["start_kp"] - keypoints["center"]
+    b = keypoints["tip"] - keypoints["center"]
+    angle2tip = cosangle(a, b)
+    # get sweep angles start->end
+    b = keypoints["end_kp"] - keypoints["center"]
+    totalAngle = cosangle(a, b)
+    return angle2tip, totalAngle
+
+
+def get_text_from_image(client, path_or_img)->Union[list[dict],Exception ]:
+    if type(path_or_img)==str:
+        with open(path_or_img, "rb") as image_file:
+            content = image_file.read()
+    else:
+        buf = io.BytesIO()
+        path_or_img.save(buf, format="JPEG")
+        content = buf.getvalue()
+
+    image = vision.Image(content=content)
+
+    response = client.text_detection(image=image)
+    if response.error.message:
+        raise Exception(
+            "{}\nFor more info on error messages, check: "
+            "https://cloud.google.com/apis/design/errors".format(response.error.message)
+        )
+
+    texts = response.text_annotations
+    contents = [ {"text": found.description, "boxCorners": [ (vert.x, vert.y) for vert in found.bounding_poly.vertices]} for found in texts]
+    return contents
+
+def median_point_of_bounding_box(x1, y1, x2, y2, x3, y3, x4, y4):
+    x_coords = [x1, x2, x3, x4]
+    y_coords = [y1, y2, y3, y4]
+    x_median = sum(x_coords) / len(x_coords)
+    y_median = sum(y_coords) / len(y_coords)
+    return x_median, y_median
+
+def to_numeric(text:str):
+    try:
+        return float(text)
+    except:
+        pass
+    return None
+
+def result_as_validvalue(contents:list[dict])->tuple[list[dict], list[str]]:
+    # only valid values and sort min to max
+    valid = []
+    other = []
+    for f in contents:
+        t = f["text"]
+        value =  to_numeric(t)
+        if "\n" in t:
+            continue
+        elif value == None and t!="":
+            other.append(t)
+            continue
+        b = f["boxCorners"]
+        m = median_point_of_bounding_box(*np.array(b).flatten())
+        valid.append({"text":f["text"], "value": value, "mid": m})
+
+    valid.sort(key=lambda e: e["value"])
+    return valid, other
+
+
+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 cosangle(a,b):
+        na = np.linalg.norm(a)
+        nb = np.linalg.norm(b)
+        ang = np.rad2deg(np.arccos(np.dot(a, b)/(na*nb)))
+        rotdir = -1 if np.cross(a,b) < 0 else 1
+        return ang , rotdir
+    # compute angles between values
+    values = [valid[0]]
+    values[0]["dang"] = 0
+    rates = []
+    angS = 0
+    for v in valid[1:]:
+        u = v.copy()
+        u["dv"] = v["value"] - values[-1]["value"]
+        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
+        angS += ang
+        u["dang"] = ang
+        u["dvda"] = u["dv"] / ang
+        rates.append(u["dvda"])
+        values.append(u)
+
+    rates = np.array(rates)
+    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, meanAng
+    return values, rate
+
+
+def vec_angle(v1, v2)->tuple[float, bool]:
+    vector1 = v1/np.linalg.norm(v1)
+    vector2 = v2/np.linalg.norm(v2)
+    angle_rad = np.arctan2(np.cross(vector1, vector2), np.dot(vector1, vector2))
+    return math.degrees(angle_rad)
+
+
+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
+
+    return values
+
+
+def get_needle_value(img, keypoints):
+
+    tic2 = time()
+    contents = get_text_from_image(client, img)
+    toc = time()
+    print(f"ocr took: {toc-tic2:.1g}")
+
+    assert len(contents)
+    valid,other = result_as_validvalue(contents)
+    assert len(valid)
+    center = np.array(keypoints["center"])
+    values, rate = determine_ocr_neighbors(center, valid)
+    assert len(values)>=2, "failed to find at least 2 OCR values"
+
+    # import pandas as pd
+    # print(pd.DataFrame.from_dict(values))
+    
+    tree = KDTree([v["mid"] for v in values])
+    # find bounding ocr values of tip
+    dist, nearestIx = tree.query(keypoints["tip"],k=2)
+    nearestIx.sort()
+    dist, nearestIx 
+
+    values = angles_from_tip(keypoints, values, nearestIx)
+    # compare against start and end
+    c = keypoints["center"]
+    tip = keypoints["tip"] - c
+
+    tipValues = []
+    for i in range(len(values)):
+        v = values[i]
+        a = v["mid"] - c
+        ang = vec_angle(a,tip)
+        before = v["before"] 
+        startValue = v["value"]
+        angle2tip = ang
+        needleVal = -1
+
+        angle2tip = before
+
+        needleVal = angle2tip * rate + startValue # tip value from nearest Ix
+        tipValues.append(needleVal)
+        print(f"{i}, {ang:.2f}, {before:.2f}, @{needleVal:.2f}, {startValue}")
+
+    # print(f"total took: {toc-tic:.1g}")
+    tipValues = np.array(tipValues)
+
+    debug(img, contents, keypoints)
+
+    startValue= float(values[0]["value"])
+    tipvalue= round(tipValues[nearestIx].mean(),2)
+    endValue= float(values[-1]["value"])
+
+    return {"startValue": startValue, "tipvalue": tipvalue, "endValue": endValue, "unitPerDeg": float(rate), "otherText": list(set(other))}
+
+
+# debug draw
+def corners2bbox(C):
+    p = np.array(C)
+    s,e = p.min(axis=0).astype(int), p.max(axis=0).astype(int)
+    return s, e
+
+def debug(img, contents, keypoints):
+    draw = ImageDraw.Draw(img)
+
+    for f in contents:
+        b = f["boxCorners"]
+        s,e = corners2bbox(b)
+        c = (255,0,0)
+        draw.rectangle((*s,*e), fill=None, outline=c, width=1)
+        m = median_point_of_bounding_box(*np.array(b).flatten())
+        draw.point(m, (255,0,255))
+    img
+
+    for v,c in zip(keypoints.values(), [(255,0,0), (0,255,0), (0,0,255),(255,0,255)]):
+        s = np.array(v)-1
+        e = np.array(v)+1
+        draw.rectangle((*s,*e), c)
+    img.save("temp-ocr.jpg")
+    print("saved debug img")
+
+
+def predict(img, detect_gauge_first):
+    if detect_gauge_first:
+        model0 = get_load_PhModel()
+        results = model0.predict(img)
+        phimg,_ = get_corners(results, img)
+    else:
+        phimg = img.copy()
+
+    model = get_load_KpModel()
+    phimg = preprocessImg(phimg)
+    results = model.predict(phimg)
+    keypoints = get_keypoints(results)
+
+    angle2tip, totalAngle = calculate_sweep_angles(keypoints)
+
+    payload = get_needle_value(phimg, keypoints)
+    payload["angleToTip"] = round(angle2tip,2)
+    payload["totalAngle"] = round(totalAngle,2)
+
+    return payload
+
+
+def test(img, detect_gauge_first):
+    return {"msg":str(img.size), "other": detect_gauge_first}
+
+
+gr.Interface(fn=predict,
+             inputs=[
+                 gr.Image(type="pil", sources=["upload","clipboard"],streaming=False, min_width=640),
+                 gr.Checkbox(True, label="detect gauge first", info="if input image is zoomed in on only one gauge, uncheck box")
+                 ],
+             outputs="json",
+             examples=[_examples],
+             cache_examples=True)\
+    .launch()

corners-best.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a88502e86a40941aec69fe4d48e03c675a9381500fbf4c1ca8e3d1a89db089a9
+size 37732202

keypoints-best.pt

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

requirements.txt

@@ -0,0 +1,8 @@
+ultralytics==8.1.2
+opencv-python==4.9.0.80
+opencv-python-headless==4.8.0.76
+numpy==1.24.1
+scipy==1.11.2
+gradio_client==0.8.0
+google-cloud-vision==3.5.0
+Pillow==9.3.0