jalan-api / src /core.py
sairusses's picture
add
b400ace
import os
from collections import defaultdict
from typing import Any
import cv2
import matplotlib.pyplot as plt
import numpy as np
import src.constants as constants
from src.logger import logger
from src.utils.image import CLAHE_HELPER, ImageUtils
from src.utils.interaction import InteractionUtils
class ImageInstanceOps:
"""Class to hold fine-tuned utilities for a group of images. One instance for each processing directory."""
save_img_list: Any = defaultdict(list)
def __init__(self, tuning_config):
super().__init__()
self.tuning_config = tuning_config
self.save_image_level = tuning_config.outputs.save_image_level
def apply_preprocessors(self, file_path, in_omr, template):
tuning_config = self.tuning_config
# resize to conform to template
in_omr = ImageUtils.resize_util(
in_omr,
tuning_config.dimensions.processing_width,
tuning_config.dimensions.processing_height,
)
# run pre_processors in sequence
for pre_processor in template.pre_processors:
in_omr = pre_processor.apply_filter(in_omr, file_path)
return in_omr
def read_omr_response(self, template, image, name, save_dir=None):
config = self.tuning_config
auto_align = config.alignment_params.auto_align
try:
img = image.copy()
# origDim = img.shape[:2]
img = ImageUtils.resize_util(
img, template.page_dimensions[0], template.page_dimensions[1]
)
if img.max() > img.min():
img = ImageUtils.normalize_util(img)
# Processing copies
transp_layer = img.copy()
final_marked = img.copy()
morph = img.copy()
self.append_save_img(3, morph)
if auto_align:
# Note: clahe is good for morphology, bad for thresholding
morph = CLAHE_HELPER.apply(morph)
self.append_save_img(3, morph)
# Remove shadows further, make columns/boxes darker (less gamma)
morph = ImageUtils.adjust_gamma(
morph, config.threshold_params.GAMMA_LOW
)
# TODO: all numbers should come from either constants or config
_, morph = cv2.threshold(morph, 220, 220, cv2.THRESH_TRUNC)
morph = ImageUtils.normalize_util(morph)
self.append_save_img(3, morph)
if config.outputs.show_image_level >= 4:
InteractionUtils.show("morph1", morph, 0, 1, config)
# Move them to data class if needed
# Overlay Transparencies
alpha = 0.65
omr_response = {}
multi_marked, multi_roll = 0, 0
# TODO Make this part useful for visualizing status checks
# blackVals=[0]
# whiteVals=[255]
if config.outputs.show_image_level >= 5:
all_c_box_vals = {"int": [], "mcq": []}
# TODO: simplify this logic
q_nums = {"int": [], "mcq": []}
# Find Shifts for the field_blocks --> Before calculating threshold!
if auto_align:
# print("Begin Alignment")
# Open : erode then dilate
v_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (2, 10))
morph_v = cv2.morphologyEx(
morph, cv2.MORPH_OPEN, v_kernel, iterations=3
)
_, morph_v = cv2.threshold(morph_v, 200, 200, cv2.THRESH_TRUNC)
morph_v = 255 - ImageUtils.normalize_util(morph_v)
if config.outputs.show_image_level >= 3:
InteractionUtils.show(
"morphed_vertical", morph_v, 0, 1, config=config
)
# InteractionUtils.show("morph1",morph,0,1,config=config)
# InteractionUtils.show("morphed_vertical",morph_v,0,1,config=config)
self.append_save_img(3, morph_v)
morph_thr = 60 # for Mobile images, 40 for scanned Images
_, morph_v = cv2.threshold(morph_v, morph_thr, 255, cv2.THRESH_BINARY)
# kernel best tuned to 5x5 now
morph_v = cv2.erode(morph_v, np.ones((5, 5), np.uint8), iterations=2)
self.append_save_img(3, morph_v)
# h_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (10, 2))
# morph_h = cv2.morphologyEx(morph, cv2.MORPH_OPEN, h_kernel, iterations=3)
# ret, morph_h = cv2.threshold(morph_h,200,200,cv2.THRESH_TRUNC)
# morph_h = 255 - normalize_util(morph_h)
# InteractionUtils.show("morph_h",morph_h,0,1,config=config)
# _, morph_h = cv2.threshold(morph_h,morph_thr,255,cv2.THRESH_BINARY)
# morph_h = cv2.erode(morph_h, np.ones((5,5),np.uint8), iterations = 2)
if config.outputs.show_image_level >= 3:
InteractionUtils.show(
"morph_thr_eroded", morph_v, 0, 1, config=config
)
self.append_save_img(6, morph_v)
# template relative alignment code
for field_block in template.field_blocks:
s, d = field_block.origin, field_block.dimensions
match_col, max_steps, align_stride, thk = map(
config.alignment_params.get,
[
"match_col",
"max_steps",
"stride",
"thickness",
],
)
shift, steps = 0, 0
while steps < max_steps:
left_mean = np.mean(
morph_v[
s[1] : s[1] + d[1],
s[0] + shift - thk : -thk + s[0] + shift + match_col,
]
)
right_mean = np.mean(
morph_v[
s[1] : s[1] + d[1],
s[0]
+ shift
- match_col
+ d[0]
+ thk : thk
+ s[0]
+ shift
+ d[0],
]
)
# For demonstration purposes-
# if(field_block.name == "int1"):
# ret = morph_v.copy()
# cv2.rectangle(ret,
# (s[0]+shift-thk,s[1]),
# (s[0]+shift+thk+d[0],s[1]+d[1]),
# constants.CLR_WHITE,
# 3)
# appendSaveImg(6,ret)
# print(shift, left_mean, right_mean)
left_shift, right_shift = left_mean > 100, right_mean > 100
if left_shift:
if right_shift:
break
else:
shift -= align_stride
else:
if right_shift:
shift += align_stride
else:
break
steps += 1
field_block.shift = shift
# print("Aligned field_block: ",field_block.name,"Corrected Shift:",
# field_block.shift,", dimensions:", field_block.dimensions,
# "origin:", field_block.origin,'\n')
# print("End Alignment")
final_align = None
if config.outputs.show_image_level >= 2:
initial_align = self.draw_template_layout(img, template, shifted=False)
final_align = self.draw_template_layout(
img, template, shifted=True, draw_qvals=True
)
# appendSaveImg(4,mean_vals)
self.append_save_img(2, initial_align)
self.append_save_img(2, final_align)
if auto_align:
final_align = np.hstack((initial_align, final_align))
self.append_save_img(5, img)
# Get mean bubbleValues n other stats
all_q_vals, all_q_strip_arrs, all_q_std_vals = [], [], []
total_q_strip_no = 0
for field_block in template.field_blocks:
box_w, box_h = field_block.bubble_dimensions
q_std_vals = []
for field_block_bubbles in field_block.traverse_bubbles:
q_strip_vals = []
for pt in field_block_bubbles:
# shifted
x, y = (pt.x + field_block.shift, pt.y)
rect = [y, y + box_h, x, x + box_w]
q_strip_vals.append(
cv2.mean(img[rect[0] : rect[1], rect[2] : rect[3]])[0]
# detectCross(img, rect) ? 100 : 0
)
q_std_vals.append(round(np.std(q_strip_vals), 2))
all_q_strip_arrs.append(q_strip_vals)
# _, _, _ = get_global_threshold(q_strip_vals, "QStrip Plot",
# plot_show=False, sort_in_plot=True)
# hist = getPlotImg()
# InteractionUtils.show("QStrip "+field_block_bubbles[0].field_label, hist, 0, 1,config=config)
all_q_vals.extend(q_strip_vals)
# print(total_q_strip_no, field_block_bubbles[0].field_label, q_std_vals[len(q_std_vals)-1])
total_q_strip_no += 1
all_q_std_vals.extend(q_std_vals)
global_std_thresh, _, _ = self.get_global_threshold(
all_q_std_vals
) # , "Q-wise Std-dev Plot", plot_show=True, sort_in_plot=True)
# plt.show()
# hist = getPlotImg()
# InteractionUtils.show("StdHist", hist, 0, 1,config=config)
# Note: Plotting takes Significant times here --> Change Plotting args
# to support show_image_level
# , "Mean Intensity Histogram",plot_show=True, sort_in_plot=True)
global_thr, _, _ = self.get_global_threshold(all_q_vals, looseness=4)
logger.info(
f"Thresholding: \tglobal_thr: {round(global_thr, 2)} \tglobal_std_THR: {round(global_std_thresh, 2)}\t{'(Looks like a Xeroxed OMR)' if (global_thr == 255) else ''}"
)
# plt.show()
# hist = getPlotImg()
# InteractionUtils.show("StdHist", hist, 0, 1,config=config)
# if(config.outputs.show_image_level>=1):
# hist = getPlotImg()
# InteractionUtils.show("Hist", hist, 0, 1,config=config)
# appendSaveImg(4,hist)
# appendSaveImg(5,hist)
# appendSaveImg(2,hist)
per_omr_threshold_avg, total_q_strip_no, total_q_box_no = 0, 0, 0
for field_block in template.field_blocks:
block_q_strip_no = 1
box_w, box_h = field_block.bubble_dimensions
shift = field_block.shift
s, d = field_block.origin, field_block.dimensions
key = field_block.name[:3]
# cv2.rectangle(final_marked,(s[0]+shift,s[1]),(s[0]+shift+d[0],
# s[1]+d[1]),CLR_BLACK,3)
for field_block_bubbles in field_block.traverse_bubbles:
# All Black or All White case
no_outliers = all_q_std_vals[total_q_strip_no] < global_std_thresh
# print(total_q_strip_no, field_block_bubbles[0].field_label,
# all_q_std_vals[total_q_strip_no], "no_outliers:", no_outliers)
per_q_strip_threshold = self.get_local_threshold(
all_q_strip_arrs[total_q_strip_no],
global_thr,
no_outliers,
f"Mean Intensity Histogram for {key}.{field_block_bubbles[0].field_label}.{block_q_strip_no}",
config.outputs.show_image_level >= 6,
)
# print(field_block_bubbles[0].field_label,key,block_q_strip_no, "THR: ",
# round(per_q_strip_threshold,2))
per_omr_threshold_avg += per_q_strip_threshold
# Note: Little debugging visualization - view the particular Qstrip
# if(
# 0
# # or "q17" in (field_block_bubbles[0].field_label)
# # or (field_block_bubbles[0].field_label+str(block_q_strip_no))=="q15"
# ):
# st, end = qStrip
# InteractionUtils.show("QStrip: "+key+"-"+str(block_q_strip_no),
# img[st[1] : end[1], st[0]+shift : end[0]+shift],0,config=config)
# TODO: get rid of total_q_box_no
detected_bubbles = []
for bubble in field_block_bubbles:
bubble_is_marked = (
per_q_strip_threshold > all_q_vals[total_q_box_no]
)
total_q_box_no += 1
if bubble_is_marked:
detected_bubbles.append(bubble)
x, y, field_value = (
bubble.x + field_block.shift,
bubble.y,
bubble.field_value,
)
cv2.rectangle(
final_marked,
(int(x + box_w / 12), int(y + box_h / 12)),
(
int(x + box_w - box_w / 12),
int(y + box_h - box_h / 12),
),
constants.CLR_DARK_GRAY,
3,
)
cv2.putText(
final_marked,
str(field_value),
(x, y),
cv2.FONT_HERSHEY_SIMPLEX,
constants.TEXT_SIZE,
(20, 20, 10),
int(1 + 3.5 * constants.TEXT_SIZE),
)
else:
cv2.rectangle(
final_marked,
(int(x + box_w / 10), int(y + box_h / 10)),
(
int(x + box_w - box_w / 10),
int(y + box_h - box_h / 10),
),
constants.CLR_GRAY,
-1,
)
for bubble in detected_bubbles:
field_label, field_value = (
bubble.field_label,
bubble.field_value,
)
# Only send rolls multi-marked in the directory
multi_marked_local = field_label in omr_response
omr_response[field_label] = (
(omr_response[field_label] + field_value)
if multi_marked_local
else field_value
)
# TODO: generalize this into identifier
# multi_roll = multi_marked_local and "Roll" in str(q)
multi_marked = multi_marked or multi_marked_local
if len(detected_bubbles) == 0:
field_label = field_block_bubbles[0].field_label
omr_response[field_label] = field_block.empty_val
if config.outputs.show_image_level >= 5:
if key in all_c_box_vals:
q_nums[key].append(f"{key[:2]}_c{str(block_q_strip_no)}")
all_c_box_vals[key].append(
all_q_strip_arrs[total_q_strip_no]
)
block_q_strip_no += 1
total_q_strip_no += 1
# /for field_block
per_omr_threshold_avg /= total_q_strip_no
per_omr_threshold_avg = round(per_omr_threshold_avg, 2)
# Translucent
cv2.addWeighted(
final_marked, alpha, transp_layer, 1 - alpha, 0, final_marked
)
# Box types
if config.outputs.show_image_level >= 6:
# plt.draw()
f, axes = plt.subplots(len(all_c_box_vals), sharey=True)
f.canvas.manager.set_window_title(name)
ctr = 0
type_name = {
"int": "Integer",
"mcq": "MCQ",
"med": "MED",
"rol": "Roll",
}
for k, boxvals in all_c_box_vals.items():
axes[ctr].title.set_text(type_name[k] + " Type")
axes[ctr].boxplot(boxvals)
# thrline=axes[ctr].axhline(per_omr_threshold_avg,color='red',ls='--')
# thrline.set_label("Average THR")
axes[ctr].set_ylabel("Intensity")
axes[ctr].set_xticklabels(q_nums[k])
# axes[ctr].legend()
ctr += 1
# imshow will do the waiting
plt.tight_layout(pad=0.5)
plt.show()
if config.outputs.show_image_level >= 3 and final_align is not None:
final_align = ImageUtils.resize_util_h(
final_align, int(config.dimensions.display_height)
)
# [final_align.shape[1],0])
InteractionUtils.show(
"Template Alignment Adjustment", final_align, 0, 0, config=config
)
if config.outputs.save_detections and save_dir is not None:
if multi_roll:
save_dir = save_dir.joinpath("_MULTI_")
image_path = str(save_dir.joinpath(name))
ImageUtils.save_img(image_path, final_marked)
self.append_save_img(2, final_marked)
if save_dir is not None:
for i in range(config.outputs.save_image_level):
self.save_image_stacks(i + 1, name, save_dir)
return omr_response, final_marked, multi_marked, multi_roll
except Exception as e:
raise e
@staticmethod
def draw_template_layout(img, template, shifted=True, draw_qvals=False, border=-1):
img = ImageUtils.resize_util(
img, template.page_dimensions[0], template.page_dimensions[1]
)
final_align = img.copy()
for field_block in template.field_blocks:
s, d = field_block.origin, field_block.dimensions
box_w, box_h = field_block.bubble_dimensions
shift = field_block.shift
if shifted:
cv2.rectangle(
final_align,
(s[0] + shift, s[1]),
(s[0] + shift + d[0], s[1] + d[1]),
constants.CLR_BLACK,
3,
)
else:
cv2.rectangle(
final_align,
(s[0], s[1]),
(s[0] + d[0], s[1] + d[1]),
constants.CLR_BLACK,
3,
)
for field_block_bubbles in field_block.traverse_bubbles:
for pt in field_block_bubbles:
x, y = (pt.x + field_block.shift, pt.y) if shifted else (pt.x, pt.y)
cv2.rectangle(
final_align,
(int(x + box_w / 10), int(y + box_h / 10)),
(int(x + box_w - box_w / 10), int(y + box_h - box_h / 10)),
constants.CLR_GRAY,
border,
)
if draw_qvals:
rect = [y, y + box_h, x, x + box_w]
cv2.putText(
final_align,
f"{int(cv2.mean(img[rect[0] : rect[1], rect[2] : rect[3]])[0])}",
(rect[2] + 2, rect[0] + (box_h * 2) // 3),
cv2.FONT_HERSHEY_SIMPLEX,
0.6,
constants.CLR_BLACK,
2,
)
if shifted:
text_in_px = cv2.getTextSize(
field_block.name, cv2.FONT_HERSHEY_SIMPLEX, constants.TEXT_SIZE, 4
)
cv2.putText(
final_align,
field_block.name,
(int(s[0] + d[0] - text_in_px[0][0]), int(s[1] - text_in_px[0][1])),
cv2.FONT_HERSHEY_SIMPLEX,
constants.TEXT_SIZE,
constants.CLR_BLACK,
4,
)
return final_align
def get_global_threshold(
self,
q_vals_orig,
plot_title=None,
plot_show=True,
sort_in_plot=True,
looseness=1,
):
"""
Note: Cannot assume qStrip has only-gray or only-white bg
(in which case there is only one jump).
So there will be either 1 or 2 jumps.
1 Jump :
......
||||||
|||||| <-- risky THR
|||||| <-- safe THR
....||||||
||||||||||
2 Jumps :
......
|||||| <-- wrong THR
....||||||
|||||||||| <-- safe THR
..||||||||||
||||||||||||
The abstract "First LARGE GAP" is perfect for this.
Current code is considering ONLY TOP 2 jumps(>= MIN_GAP) to be big,
gives the smaller one
"""
config = self.tuning_config
PAGE_TYPE_FOR_THRESHOLD, MIN_JUMP, JUMP_DELTA = map(
config.threshold_params.get,
[
"PAGE_TYPE_FOR_THRESHOLD",
"MIN_JUMP",
"JUMP_DELTA",
],
)
global_default_threshold = (
constants.GLOBAL_PAGE_THRESHOLD_WHITE
if PAGE_TYPE_FOR_THRESHOLD == "white"
else constants.GLOBAL_PAGE_THRESHOLD_BLACK
)
# Sort the Q bubbleValues
# TODO: Change var name of q_vals
q_vals = sorted(q_vals_orig)
# Find the FIRST LARGE GAP and set it as threshold:
ls = (looseness + 1) // 2
l = len(q_vals) - ls
max1, thr1 = MIN_JUMP, global_default_threshold
for i in range(ls, l):
jump = q_vals[i + ls] - q_vals[i - ls]
if jump > max1:
max1 = jump
thr1 = q_vals[i - ls] + jump / 2
# NOTE: thr2 is deprecated, thus is JUMP_DELTA
# Make use of the fact that the JUMP_DELTA(Vertical gap ofc) between
# values at detected jumps would be atleast 20
max2, thr2 = MIN_JUMP, global_default_threshold
# Requires atleast 1 gray box to be present (Roll field will ensure this)
for i in range(ls, l):
jump = q_vals[i + ls] - q_vals[i - ls]
new_thr = q_vals[i - ls] + jump / 2
if jump > max2 and abs(thr1 - new_thr) > JUMP_DELTA:
max2 = jump
thr2 = new_thr
# global_thr = min(thr1,thr2)
global_thr, j_low, j_high = thr1, thr1 - max1 // 2, thr1 + max1 // 2
# # For normal images
# thresholdRead = 116
# if(thr1 > thr2 and thr2 > thresholdRead):
# print("Note: taking safer thr line.")
# global_thr, j_low, j_high = thr2, thr2 - max2//2, thr2 + max2//2
if plot_title:
_, ax = plt.subplots()
ax.bar(range(len(q_vals_orig)), q_vals if sort_in_plot else q_vals_orig)
ax.set_title(plot_title)
thrline = ax.axhline(global_thr, color="green", ls="--", linewidth=5)
thrline.set_label("Global Threshold")
thrline = ax.axhline(thr2, color="red", ls=":", linewidth=3)
thrline.set_label("THR2 Line")
# thrline=ax.axhline(j_low,color='red',ls='-.', linewidth=3)
# thrline=ax.axhline(j_high,color='red',ls='-.', linewidth=3)
# thrline.set_label("Boundary Line")
# ax.set_ylabel("Mean Intensity")
ax.set_ylabel("Values")
ax.set_xlabel("Position")
ax.legend()
if plot_show:
plt.title(plot_title)
plt.show()
return global_thr, j_low, j_high
def get_local_threshold(
self, q_vals, global_thr, no_outliers, plot_title=None, plot_show=True
):
"""
TODO: Update this documentation too-
//No more - Assumption : Colwise background color is uniformly gray or white,
but not alternating. In this case there is atmost one jump.
0 Jump :
<-- safe THR?
.......
...|||||||
|||||||||| <-- safe THR?
// How to decide given range is above or below gray?
-> global q_vals shall absolutely help here. Just run same function
on total q_vals instead of colwise _//
How to decide it is this case of 0 jumps
1 Jump :
......
||||||
|||||| <-- risky THR
|||||| <-- safe THR
....||||||
||||||||||
"""
config = self.tuning_config
# Sort the Q bubbleValues
q_vals = sorted(q_vals)
# Small no of pts cases:
# base case: 1 or 2 pts
if len(q_vals) < 3:
thr1 = (
global_thr
if np.max(q_vals) - np.min(q_vals) < config.threshold_params.MIN_GAP
else np.mean(q_vals)
)
else:
# qmin, qmax, qmean, qstd = round(np.min(q_vals),2), round(np.max(q_vals),2),
# round(np.mean(q_vals),2), round(np.std(q_vals),2)
# GVals = [round(abs(q-qmean),2) for q in q_vals]
# gmean, gstd = round(np.mean(GVals),2), round(np.std(GVals),2)
# # DISCRETION: Pretty critical factor in reading response
# # Doesn't work well for small number of values.
# DISCRETION = 2.7 # 2.59 was closest hit, 3.0 is too far
# L2MaxGap = round(max([abs(g-gmean) for g in GVals]),2)
# if(L2MaxGap > DISCRETION*gstd):
# no_outliers = False
# # ^Stackoverflow method
# print(field_label, no_outliers,"qstd",round(np.std(q_vals),2), "gstd", gstd,
# "Gaps in gvals",sorted([round(abs(g-gmean),2) for g in GVals],reverse=True),
# '\t',round(DISCRETION*gstd,2), L2MaxGap)
# else:
# Find the LARGEST GAP and set it as threshold: //(FIRST LARGE GAP)
l = len(q_vals) - 1
max1, thr1 = config.threshold_params.MIN_JUMP, 255
for i in range(1, l):
jump = q_vals[i + 1] - q_vals[i - 1]
if jump > max1:
max1 = jump
thr1 = q_vals[i - 1] + jump / 2
# print(field_label,q_vals,max1)
confident_jump = (
config.threshold_params.MIN_JUMP
+ config.threshold_params.CONFIDENT_SURPLUS
)
# If not confident, then only take help of global_thr
if max1 < confident_jump:
if no_outliers:
# All Black or All White case
thr1 = global_thr
else:
# TODO: Low confidence parameters here
pass
# if(thr1 == 255):
# print("Warning: threshold is unexpectedly 255! (Outlier Delta issue?)",plot_title)
# Make a common plot function to show local and global thresholds
if plot_show and plot_title is not None:
_, ax = plt.subplots()
ax.bar(range(len(q_vals)), q_vals)
thrline = ax.axhline(thr1, color="green", ls=("-."), linewidth=3)
thrline.set_label("Local Threshold")
thrline = ax.axhline(global_thr, color="red", ls=":", linewidth=5)
thrline.set_label("Global Threshold")
ax.set_title(plot_title)
ax.set_ylabel("Bubble Mean Intensity")
ax.set_xlabel("Bubble Number(sorted)")
ax.legend()
# TODO append QStrip to this plot-
# appendSaveImg(6,getPlotImg())
if plot_show:
plt.show()
return thr1
def append_save_img(self, key, img):
if self.save_image_level >= int(key):
self.save_img_list[key].append(img.copy())
def save_image_stacks(self, key, filename, save_dir):
config = self.tuning_config
if self.save_image_level >= int(key) and self.save_img_list[key] != []:
name = os.path.splitext(filename)[0]
result = np.hstack(
tuple(
[
ImageUtils.resize_util_h(img, config.dimensions.display_height)
for img in self.save_img_list[key]
]
)
)
result = ImageUtils.resize_util(
result,
min(
len(self.save_img_list[key]) * config.dimensions.display_width // 3,
int(config.dimensions.display_width * 2.5),
),
)
ImageUtils.save_img(f"{save_dir}stack/{name}_{str(key)}_stack.jpg", result)
def reset_all_save_img(self):
for i in range(self.save_image_level):
self.save_img_list[i + 1] = []