Download tinydoc/pipeline/detect.py from harryfrz/tinydoc-5m: direct link, hf CLI and curl.
- Browser
- Download file 8.7 kB
-
https://huggingface.co/harryfrz/tinydoc-5m/resolve/main/tinydoc/pipeline/detect.py
- Command line
-
hf download hf://harryfrz/tinydoc-5m/tinydoc/pipeline/detect.py
-
curl -L -o detect.py https://huggingface.co/harryfrz/tinydoc-5m/resolve/main/tinydoc/pipeline/detect.py
8.7 kB
| """Detector pre/post-processing shared by evaluation and inference.""" | |
| import cv2 | |
| import numpy as np | |
| from ..geometry import expand_quad, quad_size | |
| from ..models.detector import KERNEL_END_TRIM, KERNEL_HEIGHT_RATIO | |
| DET_MAX_SIDE = 1920 | |
| DET_MIN_SIDE = 960 | |
| def detector_input(img_rgb, max_side=DET_MAX_SIDE, min_side=DET_MIN_SIDE): | |
| """Resize so the long side is within [min_side, max_side], pad to a multiple of 32. | |
| Returns (float32 CHW in [0,1], scale) where model coords = original coords * scale. | |
| """ | |
| h, w = img_rgb.shape[:2] | |
| long = max(h, w) | |
| scale = min(max_side / long, max(1.0, min_side / long)) | |
| nh, nw = max(32, round(h * scale)), max(32, round(w * scale)) | |
| img = cv2.resize(img_rgb, (nw, nh), interpolation=cv2.INTER_AREA if scale < 1 else cv2.INTER_LINEAR) | |
| ph, pw = (nh + 31) // 32 * 32, (nw + 31) // 32 * 32 | |
| pad = np.empty((ph, pw, 3), np.uint8) | |
| pad[:] = np.median(img[::8, ::8].reshape(-1, 3), 0).astype(np.uint8) # page colour from a pixel sample | |
| pad[:nh, :nw] = img | |
| return pad.transpose(2, 0, 1).astype(np.float32) / 255.0, scale | |
| def _text_quad(rect): | |
| """minAreaRect -> quad ordered TL, TR, BR, BL with the top edge as close to horizontal as possible.""" | |
| pts = cv2.boxPoints(rect) | |
| c = pts.mean(0) | |
| pts = pts[np.argsort(np.arctan2(pts[:, 1] - c[1], pts[:, 0] - c[0]))] # clockwise in image coords | |
| best, best_score = 0, -1e9 | |
| for s in range(4): | |
| v = pts[(s + 1) % 4] - pts[s] | |
| score = v[0] / (np.linalg.norm(v) + 1e-6) # cos of angle to +x | |
| if score > best_score: | |
| best, best_score = s, score | |
| return np.roll(pts, -best, axis=0).astype(np.float32) | |
| def kernels_to_quads(prob, bin_thresh=0.3, box_thresh=0.55, min_area=4): | |
| """prob: (H, W) kernel probability map. Returns list of (quad, score) in map coordinates.""" | |
| bitmap = (prob > bin_thresh).astype(np.uint8) | |
| n, labels, stats, _ = cv2.connectedComponentsWithStats(bitmap, connectivity=4) | |
| out = [] | |
| for i in range(1, n): | |
| x, y, w, h, area = stats[i] | |
| if area < min_area: | |
| continue | |
| comp = labels[y:y + h, x:x + w] == i | |
| score = float(prob[y:y + h, x:x + w][comp].mean()) | |
| if score < box_thresh: | |
| continue | |
| ys, xs = np.nonzero(comp) | |
| pts = np.stack([xs + x, ys + y], 1).astype(np.float32) | |
| (cx, cy), (rw, rh), ang = cv2.minAreaRect(pts) | |
| q = _text_quad(((cx, cy), (rw, rh), ang)) # fillPoly includes boundary pixels, so centres already span the kernel | |
| kw, kh = quad_size(q) | |
| full_h = kh / KERNEL_HEIGHT_RATIO | |
| dy = (full_h - kh) / 2 | |
| dx = min(KERNEL_END_TRIM * full_h, kw / 2) | |
| out.append((expand_quad(q, dx, dy, dx, dy), score)) | |
| return out | |
| def quad_iou(a, b): | |
| a, b = np.asarray(a, np.float32), np.asarray(b, np.float32) | |
| inter, _ = cv2.intersectConvexConvex(a, b) | |
| union = abs(cv2.contourArea(a)) + abs(cv2.contourArea(b)) - inter | |
| return float(inter / union) if union > 0 else 0.0 | |
| def match_quads(pred, gt, thresh=0.5): | |
| """Greedy one-to-one matching by IoU. Returns (matches [(pi, gi, iou)], unmatched_pred, unmatched_gt).""" | |
| pairs = [] | |
| for i, p in enumerate(pred): | |
| pb = cv2.boundingRect(np.asarray(p, np.float32)) | |
| for j, g in enumerate(gt): | |
| gb = cv2.boundingRect(np.asarray(g, np.float32)) | |
| if pb[0] > gb[0] + gb[2] or gb[0] > pb[0] + pb[2] or pb[1] > gb[1] + gb[3] or gb[1] > pb[1] + pb[3]: | |
| continue | |
| iou = quad_iou(p, g) | |
| if iou >= thresh: | |
| pairs.append((iou, i, j)) | |
| pairs.sort(reverse=True) | |
| used_p, used_g, matches = set(), set(), [] | |
| for iou, i, j in pairs: | |
| if i not in used_p and j not in used_g: | |
| used_p.add(i) | |
| used_g.add(j) | |
| matches.append((i, j, iou)) | |
| return matches, [i for i in range(len(pred)) if i not in used_p], [j for j in range(len(gt)) if j not in used_g] | |
| def find_rules(img, text_height): | |
| """Printed horizontal rules (thin, long dark lines) as [x0, y0, x1, y1] boxes. | |
| Newspapers and magazines separate stories with such rules; reading order uses them as separators. | |
| Text survives the long horizontal opening only as underlines, which are harmless separators. | |
| """ | |
| gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY) if img.ndim == 3 else img | |
| h, w = gray.shape | |
| ink = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_MEAN_C, cv2.THRESH_BINARY_INV, 31, 12) | |
| length = max(30, int(max(8 * text_height, w / 25))) | |
| lines = _long_runs(ink, length) | |
| lines = cv2.dilate(lines, cv2.getStructuringElement(cv2.MORPH_RECT, (max(3, int(text_height)), 1))) # bridge dashes | |
| n, _, stats, _ = cv2.connectedComponentsWithStats(lines, connectivity=8) | |
| out = [] | |
| for x, y, bw, bh, _ in stats[1:n]: | |
| if bw >= length and bh <= max(4, 0.5 * text_height): | |
| out.append([float(x), float(y), float(x + bw), float(y + bh)]) | |
| return out | |
| def _long_runs(ink, length): | |
| """Binary opening with a 1 x length horizontal line, computed from run lengths: keeps exactly the horizontal | |
| ink runs at least `length` long (several times faster than the sliding-window morphology on full pages).""" | |
| b = np.zeros((ink.shape[0], ink.shape[1] + 2), np.int8) | |
| b[:, 1:-1] = ink > 0 | |
| d = np.diff(b, axis=1) | |
| ys, x0 = np.nonzero(d == 1) | |
| _, x1 = np.nonzero(d == -1) | |
| keep = np.nonzero(x1 - x0 >= length)[0] | |
| out = np.zeros_like(ink) | |
| for k in keep: | |
| out[ys[k], x0[k]:x1[k]] = 255 | |
| return out | |
| def split_gutter_crossers(gray, dets): | |
| """Split detected lines that run across a column gutter (two columns' lines merged by the detector). | |
| A line is split only where (a) prose lines just above and below it leave a clear vertical gap - a | |
| column gutter - on both sides, (b) the line itself has whitespace there, and (c) it is body-text sized | |
| (headlines legitimately span columns). dets: list of (quad, score); returns the same. | |
| """ | |
| if len(dets) < 6: | |
| return dets | |
| boxes = np.array([[q[:, 0].min(), q[:, 1].min(), q[:, 0].max(), q[:, 1].max()] for q, _ in dets], np.float64) | |
| hs = boxes[:, 3] - boxes[:, 1] | |
| hmed = float(np.median(hs)) | |
| cy = (boxes[:, 1] + boxes[:, 3]) / 2 | |
| out = [] | |
| for k, (q, score) in enumerate(dets): | |
| x0, y0, x1, y1 = boxes[k] | |
| if not (0.7 * hmed < hs[k] < 1.4 * hmed and x1 - x0 > 6 * hmed): | |
| out.append((q, score)) | |
| continue | |
| near = (np.abs(cy - cy[k]) < 6 * hmed) & (boxes[:, 2] > x0) & (boxes[:, 0] < x1) | |
| near[k] = False | |
| if near.sum() < 4: | |
| out.append((q, score)) | |
| continue | |
| lo = int(np.floor(x0)) | |
| cov = np.zeros(int(np.ceil(x1)) - lo + 1, np.int32) | |
| for bx0, _, bx1, _ in boxes[near]: | |
| cov[max(0, int(bx0) - lo):max(0, int(bx1) - lo)] += 1 | |
| split_at = None | |
| free = np.flatnonzero(np.diff(np.r_[1, (cov == 0).astype(np.int8), 1])) # boundaries of zero runs | |
| for a, b in zip(free[0::2], free[1::2]): | |
| if b - a < 0.3 * hmed or a < hmed or b > len(cov) - hmed: | |
| continue # too narrow, or at the line's ends | |
| gx0, gx1 = lo + a, lo + b | |
| widths = boxes[:, 2] - boxes[:, 0] | |
| ok = True | |
| for side in (near & (boxes[:, 2] <= gx0 + 1), near & (boxes[:, 0] >= gx1 - 1)): | |
| # a real column gutter has prose lines on both sides, above and below this line | |
| # (a table's column gaps under a caption are not one) | |
| above, below = side & (cy < cy[k]), side & (cy > cy[k]) | |
| if above.sum() < 1 or below.sum() < 1 or np.median(widths[side]) < 6 * hmed: | |
| ok = False | |
| if not ok: | |
| continue | |
| crop = gray[max(0, int(y0)):int(np.ceil(y1)), max(0, int(gx0 - 0.3 * hmed)):int(np.ceil(gx1 + 0.3 * hmed))] | |
| if crop.size == 0: | |
| continue | |
| thr = min(200, int(np.median(crop)) - 40) | |
| empty = (crop < thr).sum(0) == 0 | |
| runs = np.flatnonzero(np.diff(np.r_[0, empty.astype(np.int8), 0])) | |
| best = max(((e - s_, s_) for s_, e in zip(runs[0::2], runs[1::2])), default=(0, 0)) | |
| if best[0] >= 0.25 * hmed: | |
| split_at = int(gx0 - 0.3 * hmed) + best[1] + best[0] / 2 | |
| break | |
| if split_at is None: | |
| out.append((q, score)) | |
| continue | |
| for a, b in ((x0, split_at - 1), (split_at + 1, x1)): | |
| out.append((np.array([[a, y0], [b, y0], [b, y1], [a, y1]], np.float32), score)) | |
| return out | |