import mmcv import itertools import numpy as np import cv2 from skimage.morphology import skeletonize from scipy.interpolate import CubicSpline, interp1d import matplotlib.pyplot as plt from mmdet.apis import (inference_detector, init_detector) # import sys # sys.path.append("/home/csgrad/jayashok/Mask2Former/") from clean_chart import get_clean_input import line_utils import copy def hsv_to_bgr(h, s, v): # Get RGB values c = v * s x = c * (1 - abs((h * 6) % 2 - 1)) m = v - c if h < 1/6: r, g, b = c, x, 0 elif h < 1/3: r, g, b = x, c, 0 elif h < 0.5: r, g, b = 0, c, x elif h < 2/3: r, g, b = 0, x, c elif h < 5/6: r, g, b = x, 0, c else: r, g, b = c, 0, x # Scale RGB values to 0-255 range and convert to integers r = int((r + m) * 255) g = int((g + m) * 255) b = int((b + m) * 255) return (b, g, r) def get_distinct_colors(n): huePartition = 1.0 / (n + 1) return (hsv_to_bgr(huePartition * value, 1.0, 1.0) for value in range(0, n)) def load_model(config, ckpt, device): global model model = init_detector(config, ckpt, device=device) return def do_instance(model, img, score_thr=0.3): # test a single image result = inference_detector(model, img) return parse_result(result, score_thr) def parse_result(result, score_thresh=0.3): line_data = result # print(type(result)) bbox, masks = line_data[0][0], line_data[1][0] inst_masks = list(itertools.compress(masks, ((bbox[:, 4] > score_thresh).tolist()))) return inst_masks def draw_lines(img, masks): annot_img = img.copy() colors = list(get_distinct_colors(len(masks))) color_map = dict(zip(range(len(masks)), colors)) # show_img(img, is_bgr=True, title='original') for idx, mask in enumerate(masks): annot_img[mask] = color_map[idx] # show_img(img, is_bgr=True, title=f'line_{idx+1}') return annot_img def connect_lines(img): #img = cv2.imread('line_join_test2.png', 0) # grayscale image #img1 = cv2.imread('line_join_test2.png', 1) # color image th = cv2.threshold(img.astype(np.uint8), 150, 255, cv2.THRESH_BINARY)[1] kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (10, 10)) #(19, 19) img = cv2.morphologyEx(th, cv2.MORPH_DILATE, kernel) cnts1 = cv2.findContours(img, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) cnts = cnts1[0] # storing contours in a variable for i in range(len(cnts)): min_dist = max(img.shape[0], img.shape[1]) cl = [] ci = cnts[i] ci_left = tuple(ci[ci[:, :, 0].argmin()][0]) ci_right = tuple(ci[ci[:, :, 0].argmax()][0]) ci_top = tuple(ci[ci[:, :, 1].argmin()][0]) ci_bottom = tuple(ci[ci[:, :, 1].argmax()][0]) ci_list = [ci_bottom, ci_left, ci_right, ci_top] for j in range(i + 1, len(cnts)): cj = cnts[j] cj_left = tuple(cj[cj[:, :, 0].argmin()][0]) cj_right = tuple(cj[cj[:, :, 0].argmax()][0]) cj_top = tuple(cj[cj[:, :, 1].argmin()][0]) cj_bottom = tuple(cj[cj[:, :, 1].argmax()][0]) cj_list = [cj_bottom, cj_left, cj_right, cj_top] for pt1 in ci_list: for pt2 in cj_list: dist = int(np.linalg.norm(np.array(pt1) - np.array(pt2))) if dist < min_dist: min_dist = dist cl = [] cl.append([pt1, pt2, min_dist]) if len(cl) > 0: cv2.line(img, cl[0][0], cl[0][1], (255, 255, 255), thickness = 5) img = img//255 img = skeletonize(img).astype(np.uint8) img = img * 255 return img def interpolate(line_ds, inter_type='linear'): """ pred_ds: predicted data series inter_type: type of interpolation linear or cubic_spline returns list of interpolation objects for each line in the mask """ x = [] y = [] for pt in line_ds: x.append(pt['x']) y.append(pt['y']) # Remove duplicates unique_x = [] unique_y = [] for i in range(len(x)): if x.count(x[i]) == 1: unique_x.append(int(x[i])) unique_y.append(int(y[i])) if len(unique_x) < 2: return line_ds # Interpolate if inter_type == 'linear': inter = interp1d(unique_x, unique_y) if inter_type == 'cubic_spline': inter = CubicSpline(unique_x, unique_y) inter_line_ds = [] x_min = min(unique_x) x_max = max(unique_x) for x in range(x_min, x_max+1): inter_line_ds.append({"x":x, "y":int(inter(x))}) return inter_line_ds def post_process(inst_masks): post_process_mask = [] for i in range(len(inst_masks)): post_process_mask.append(connect_lines(inst_masks[i])) return post_process_mask def rescale_pred_ds(ds, transformation): ds = copy.deepcopy(ds) (sx, sy, tx_crop, ty_crop, tx_padd, ty_padd) = transformation # print(sx, sy, tx_crop, ty_crop, tx_padd, ty_padd) for ln in ds: for pt in ln: pt['x'] = int((pt['x']-tx_padd) / sx) + tx_crop pt['y'] = int((pt['y']-ty_padd) / sy) + ty_crop return ds def get_dataseries(img, annot=None, to_clean=False, post_proc=False, return_masks=False): """ img: chart image as numpy array (3 channel) annot: json annot object in PMC format (required for cleaning the chart image before data extraction) mask_kp_sample_interval: interval to sample points from predicted line mask to get data series returns data series in pmc task 6a format ('visual elements') => list of lines, each a list of {x:, y: } points w.r.t original image """ global model # clean the image # save the transformation for clean image if to_clean: clean_img, transformation = get_clean_input(img, annot) else: clean_img = img # Image.fromarray(clean_img) # get inference masks inst_masks = do_instance(model, clean_img, score_thr=0.3) # return inst_masks # mask_thresh = 0.5 inst_masks = [line_mask.astype(np.uint8)*255 for line_mask in inst_masks] # for line_masks in inst_masks: # line_masks[:, line_masks.sum(axis=1)>1] = 0 # plt.imshow(inst_masks[0]) # plt.show() # return inst_masks if post_proc: inst_masks = post_process(inst_masks) # annot_img = infer.draw_lines(clean_img, inst_masks) # plt.imshow(annot_img) # inference data series pred_ds = [] for line_mask in inst_masks: ys, xs = np.nonzero(line_mask) if len(xs) == 0: pred_ds.append([]) continue x_to_ys = {} for x, y in zip(xs, ys): x_to_ys.setdefault(x, []).append(y) sorted_xs = sorted(x_to_ys.keys()) centerline = [{"x": x, "y": int(np.median(x_to_ys[x]))} for x in sorted_xs] pred_ds.append(centerline) # Reverse that transformation on pred-ds if to_clean: pred_ds = rescale_pred_ds(pred_ds, transformation) if return_masks: return pred_ds, inst_masks else: return pred_ds # Swin Transformer Backbone #CONFIG = "lineformer_swin_t_config.py" #CKPT = "train_work_dir/best_segm_mAP_iter_500.pth" #DEVICE = 'cpu' #load_model(CONFIG, CKPT, DEVICE) #print('Loaded Model:', CKPT) # if __name__ == '__main__': # img_path = "/a2il/data/ChartAnalysis/pmc_2020_split4/val_images/PMC3169544___pgen.1002274.g005.png" # #Note: Image is Loaded as BGR to RGB # img = mmcv.imread(img_path) # inst_masks = do_instance(model, img, score_thr=0.3) # annot_img = draw_lines(img, inst_masks) # post_processed_mask = post_process(inst_masks) # for i in range(len(post_processed_mask)): # cv2.imwrite(str(i)+".jpg", post_processed_mask[i]) # #cv2.imwrite(str(i)+".jpg", inst_masks[i].astype(np.uint8)*255)