erikandClaude Opus 5.5
2ef567580a
RapidOCR's angle classifier judges each cut-out line on its own and now
and then turns a perfectly upright line upside down, which then reads as
garbage ("Jo a ds ne t i d s t d no" for a line ending in "or" on the
latest scan). It is off now. All lines on a page face the same way, so a
page that reads poorly (upright text scores 0.97-1.0 per character,
upside-down text 0.55-0.8) is read again rotated 180 degrees and the
better reading kept; that also covers pages turned 90 degrees. The app
shows an upside-down page turned the right way up.
Upright pages read about 25% faster without the per-line classifier.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""OCR of scanned pages with RapidOCR (PaddleOCR's PP-OCR models on ONNX Runtime).
|
||
|
||
pip-installable on Linux, Windows and macOS with no system packages; the default
|
||
multilingual PP-OCRv6 models ship inside the wheel, so it works offline. They read
|
||
Latin script (English, Swedish, German, French, ...) as well as Chinese and Japanese.
|
||
Independent of the scanner hardware, like myscan.stitching.
|
||
|
||
lines = Ocr()(bgr) # [Line(box, text, score, edges)], box = 4x2 corner points
|
||
lines, flipped = Ocr().read(bgr) # ... and whether the page was upside down
|
||
text, spans = layout(lines) # plain text; spans[i] = (start, end) of lines[i] in it
|
||
joins = paragraph_joins(lines, text, spans)
|
||
reflow(text, joins) # the text with paragraphs on one line each
|
||
|
||
Each line knows where its characters are (from the recogniser's per-character boxes),
|
||
so text can be selected on the image: Line.locate maps a point to a character
|
||
position, Line.quad gives the outline of a range of characters.
|
||
"""
|
||
import re
|
||
import threading
|
||
from dataclasses import dataclass
|
||
|
||
import numpy as np
|
||
|
||
|
||
@dataclass
|
||
class Line:
|
||
box: np.ndarray # 4x2 float corners (clockwise from top left), image pixels
|
||
text: str
|
||
score: float
|
||
edges: np.ndarray = None # len(text) + 1 character boundaries, 0..1 along the line
|
||
|
||
def __post_init__(self):
|
||
if self.edges is None: # no character boxes: assume evenly spaced characters
|
||
self.edges = np.linspace(0.0, 1.0, len(self.text) + 1)
|
||
|
||
@property
|
||
def width(self):
|
||
"""Length along the box's top edge."""
|
||
return float(np.hypot(*(self.box[1] - self.box[0])))
|
||
|
||
@property
|
||
def height(self):
|
||
"""Text height along the box's left edge (not the axis-aligned extent, which
|
||
grows with the slant of a skewed page)."""
|
||
return float(np.hypot(*(self.box[3] - self.box[0])))
|
||
|
||
def local(self, p):
|
||
"""Image point -> (u, v): u runs 0..1 along the line, v 0..1 from its top to
|
||
its bottom edge."""
|
||
b = self.box
|
||
m = np.array([b[1] - b[0], b[3] - b[0]]).T
|
||
return np.linalg.solve(m, np.asarray(p, float) - b[0])
|
||
|
||
def locate(self, p):
|
||
"""(distance in pixels from the line's box, character position nearest p)."""
|
||
u, v = self.local(p)
|
||
b = self.box
|
||
du = max(-u, u - 1, 0) * np.hypot(*(b[1] - b[0]))
|
||
dv = max(-v, v - 1, 0) * self.height
|
||
return float(np.hypot(du, dv)), int(np.abs(self.edges - u).argmin())
|
||
|
||
def quad(self, a, b):
|
||
"""4x2 outline of characters a..b (positions in text)."""
|
||
top, bot = self.box[[0, 1]], self.box[[3, 2]]
|
||
ta, tb = self.edges[a], self.edges[b]
|
||
lerp = lambda e, t: e[0] + t * (e[1] - e[0])
|
||
return np.array([lerp(top, ta), lerp(top, tb), lerp(bot, tb), lerp(bot, ta)])
|
||
|
||
def word(self, k):
|
||
"""(start, end) of the word around character position k."""
|
||
a = b = min(k, len(self.text))
|
||
while a > 0 and not self.text[a - 1].isspace():
|
||
a -= 1
|
||
while b < len(self.text) and not self.text[b].isspace():
|
||
b += 1
|
||
return a, b
|
||
|
||
|
||
def char_edges(box, text, chars):
|
||
"""Character boundaries along a line (0..1) from the recogniser's boxes for its
|
||
non-space characters (4x2 corners each); a space spans the gap between its
|
||
neighbours. None if the boxes don't match the text."""
|
||
if not chars or len(chars) != len(text) - text.count(" "):
|
||
return None
|
||
m0, m1 = (box[0] + box[3]) / 2, (box[1] + box[2]) / 2
|
||
d = m1 - m0
|
||
t = lambda p: float(np.dot(p - m0, d) / np.dot(d, d))
|
||
ext = iter([(t((c[0] + c[3]) / 2), t((c[1] + c[2]) / 2)) for c in (np.asarray(c, float) for c in chars)])
|
||
lo, hi = [None] * len(text), [None] * len(text)
|
||
for i, ch in enumerate(text):
|
||
if ch != " ":
|
||
lo[i], hi[i] = next(ext)
|
||
end = 0.0
|
||
for i in range(len(text)): # spaces start where the previous character ends
|
||
if lo[i] is None:
|
||
lo[i] = end
|
||
else:
|
||
end = hi[i]
|
||
start = 1.0
|
||
for i in reversed(range(len(text))): # ... and end where the next one starts
|
||
if hi[i] is None:
|
||
hi[i] = start
|
||
else:
|
||
start = lo[i]
|
||
edges = [0.0] + [(hi[i - 1] + lo[i]) / 2 for i in range(1, len(text))] + [1.0]
|
||
return np.clip(np.maximum.accumulate(edges), 0.0, 1.0)
|
||
|
||
|
||
def quality(lines):
|
||
"""Mean confidence per character."""
|
||
n = sum(len(ln.text) for ln in lines)
|
||
return sum(ln.score * len(ln.text) for ln in lines) / n if n else 0.0
|
||
|
||
|
||
class Ocr:
|
||
"""Lazily loads the engine (about 0.3 s) on first use; calls are serialised.
|
||
|
||
RapidOCR's angle classifier, which flips each line it thinks is upside down, is off:
|
||
it judges lines one by one and now and then turns a perfectly upright line over,
|
||
reading it as garbage. All lines on a page face the same way, so the page is turned
|
||
as a whole instead: upright text reads at 0.97-1.0 confidence and upside-down text
|
||
at 0.55-0.8, so a page reading poorly is read again rotated 180 degrees, and the
|
||
better reading kept. That also covers pages turned 90 degrees, whose lines are cut
|
||
out turned one way, which is either right or upside down."""
|
||
|
||
def __init__(self, **params):
|
||
self.params = {"Global.log_level": "warning", "Global.use_cls": False, **params}
|
||
self._engine, self._lock = None, threading.Lock()
|
||
|
||
def engine(self):
|
||
if self._engine is None:
|
||
from rapidocr import RapidOCR
|
||
self._engine = RapidOCR(params=self.params)
|
||
return self._engine
|
||
|
||
def __call__(self, bgr):
|
||
return self.read(bgr)[0]
|
||
|
||
def read(self, bgr):
|
||
"""(lines, flipped): flipped if the page read better rotated 180 degrees, in
|
||
which case the boxes are in the coordinates of the rotated image."""
|
||
lines = self.lines(bgr)
|
||
if quality(lines) < 0.9:
|
||
turned = self.lines(np.ascontiguousarray(bgr[::-1, ::-1]))
|
||
if quality(turned) >= max(0.85, quality(lines)):
|
||
return turned, True
|
||
return lines, False
|
||
|
||
def lines(self, bgr):
|
||
with self._lock:
|
||
r = self.engine()(bgr, return_word_box=True, return_single_char_box=True)
|
||
if r.boxes is None:
|
||
return []
|
||
words = r.word_results or [None] * len(r.txts)
|
||
out = []
|
||
for b, t, s, w in zip(r.boxes, r.txts, r.scores, words):
|
||
b = np.asarray(b, float)
|
||
out.append(Line(b, t, float(s), char_edges(b, t, [c[2] for c in w]) if w else None))
|
||
return out
|
||
|
||
|
||
def text_direction(lines):
|
||
"""Unit vector along the text lines: the length-weighted median direction of the
|
||
lines' top edges (short lines, whose angle is unreliable, only if there is nothing
|
||
else). A scan held at an angle has every line slanted the same way."""
|
||
if not lines:
|
||
return np.array([1.0, 0.0])
|
||
tops = np.array([ln.box[1] - ln.box[0] for ln in lines])
|
||
length = np.hypot(tops[:, 0], tops[:, 1])
|
||
long = length > 3 * np.array([ln.height for ln in lines])
|
||
if long.any():
|
||
tops, length = tops[long], length[long]
|
||
angle = np.arctan2(tops[:, 1], tops[:, 0])
|
||
order = np.argsort(angle)
|
||
cum = np.cumsum(length[order])
|
||
a = angle[order][np.searchsorted(cum, cum[-1] / 2)]
|
||
return np.array([np.cos(a), np.sin(a)])
|
||
|
||
|
||
def page_coords(lines):
|
||
"""(along, across) position of each line in the page's own frame, unaffected by the
|
||
scan's slant: where the line starts along the text direction, and the offset of its
|
||
centre perpendicular to it (the same anywhere along a straight line)."""
|
||
d = text_direction(lines)
|
||
n = np.array([-d[1], d[0]])
|
||
return [(float((ln.box @ d).min()), float(ln.box.mean(0) @ n)) for ln in lines]
|
||
|
||
|
||
def rows(lines):
|
||
"""Group lines into visual rows (top to bottom), each sorted left to right, in the
|
||
page's frame (page_coords). A line joins a row when its centre is within half a line
|
||
height of the row's. Returns (rows, the rows' centre offsets)."""
|
||
pos = {id(ln): p for ln, p in zip(lines, page_coords(lines))}
|
||
across = lambda ln: pos[id(ln)][1]
|
||
out = []
|
||
for ln in sorted(lines, key=across):
|
||
for row in out[-3:]:
|
||
if abs(across(ln) - np.mean([across(r) for r in row])) < 0.5 * min(ln.height, row[0].height):
|
||
row.append(ln)
|
||
break
|
||
else:
|
||
out.append([ln])
|
||
return ([sorted(row, key=lambda ln: pos[id(ln)][0]) for row in out],
|
||
[float(np.mean([across(ln) for ln in row])) for row in out])
|
||
|
||
|
||
def layout(lines):
|
||
"""Reading-order text: rows joined by newlines, with a blank line where the step to
|
||
the next row is over 1.5 times the typical line pitch (a new paragraph); items in a
|
||
row are separated by a tab. Returns (text, spans), spans[i] = (start, end) of
|
||
lines[i] in text."""
|
||
index = {id(ln): i for i, ln in enumerate(lines)}
|
||
rs, cys = rows(lines)
|
||
steps = np.diff(cys)
|
||
pitch = float(np.median(steps)) if len(steps) else 0.0
|
||
spans, parts, pos = [None] * len(lines), [], 0
|
||
for k, row in enumerate(rs):
|
||
if k:
|
||
sep = "\n\n" if steps[k - 1] > 1.5 * max(pitch, 0.6 * row[0].height) else "\n"
|
||
parts.append(sep)
|
||
pos += len(sep)
|
||
for j, ln in enumerate(row):
|
||
if j:
|
||
parts.append("\t")
|
||
pos += 1
|
||
spans[index[id(ln)]] = (pos, pos + len(ln.text))
|
||
parts.append(ln.text)
|
||
pos += len(ln.text)
|
||
return "".join(parts), spans
|
||
|
||
|
||
def _extent(ln, o, d):
|
||
"""(start, end) of a line's box along direction d (unit vector) from point o."""
|
||
t = (ln.box - o) @ d
|
||
return float(t.min()), float(t.max())
|
||
|
||
|
||
# a list item's marker: "(a)", "a)", "1.", "iv)", a bullet or dash, then a space
|
||
MARKER = re.compile(r"(\(?[0-9A-Za-z]{1,3}[.)]|[•·▪◦*\-–])\s+")
|
||
|
||
|
||
def _continues(rs, k):
|
||
"""Whether the first line of row k + 1 continues the paragraph of the last line of row
|
||
k: similar text height, the next line isn't indented (except to line up with the
|
||
text after a list marker, a hanging indent), and line k is full, i.e. the next
|
||
line's first word would not have fitted in the space left at its end (which works
|
||
for justified and ragged-right text alike)."""
|
||
a, b = rs[k][-1], rs[k + 1][0]
|
||
h = min(a.height, b.height)
|
||
if not a.text.strip() or not b.text.strip() or abs(a.height - b.height) > 0.3 * max(a.height, b.height):
|
||
return False
|
||
o = a.box[0]
|
||
d = (a.box[1] - a.box[0]) / a.width
|
||
la, ra = _extent(a, o, d)
|
||
lb, _ = _extent(b, o, d)
|
||
if lb > la + 0.8 * h:
|
||
m = MARKER.match(a.text)
|
||
if not m or abs(lb - float((a.quad(m.end(), m.end())[0] - o) @ d)) > 0.8 * h:
|
||
return False
|
||
left = min(la, lb)
|
||
right = ra
|
||
for j in range(max(k - 2, 0), min(k + 4, len(rs))): # the paragraph's right margin
|
||
c = rs[j][-1]
|
||
if abs(c.height - h) < 0.3 * h:
|
||
lj, rj = _extent(c, o, d)
|
||
if abs(lj - left) < 3 * h:
|
||
right = max(right, rj)
|
||
_, e = b.word(0)
|
||
word = (b.edges[e] - b.edges[0]) * b.width
|
||
return right - ra < word + 0.5 * h
|
||
|
||
|
||
def paragraph_joins(lines, text, spans):
|
||
"""Where layout()'s text breaks a paragraph across lines, as edits for reflow():
|
||
(start, end, replacement) replacing a line break with a space, or removing it along
|
||
with the hyphen of a word split over two lines (the hyphen stays, without the
|
||
break, when the next line starts with a capital, as in "Nord-Amerika"). Only a row
|
||
holding a single line continues a paragraph, and only a single line or a wide last
|
||
item (as after a separately read section number) runs on into it; rows of several
|
||
short items are a table."""
|
||
index = {id(ln): i for i, ln in enumerate(lines)}
|
||
rs, _ = rows(lines)
|
||
out = []
|
||
for k in range(len(rs) - 1):
|
||
a, b = rs[k][-1], rs[k + 1][0]
|
||
if len(rs[k + 1]) != 1 or (len(rs[k]) != 1 and a.width < 10 * a.height):
|
||
continue
|
||
e = spans[index[id(a)]][1]
|
||
if text[e:spans[index[id(b)]][0]] != "\n" or not _continues(rs, k):
|
||
continue
|
||
if len(a.text) > 1 and a.text[-1] in "-\u00ad¬" and a.text[-2].isalpha():
|
||
out.append((e - 1, e + 1, "") if b.text[0].islower() else (e, e + 1, ""))
|
||
else:
|
||
out.append((e, e + 1, " "))
|
||
return out
|
||
|
||
|
||
def reflow(text, joins, a=0, b=None):
|
||
"""text[a:b] with the paragraph_joins() edits that lie inside it applied."""
|
||
b = len(text) if b is None else b
|
||
parts, pos = [], a
|
||
for s, e, r in joins:
|
||
if a <= s and e <= b:
|
||
parts += [text[pos:s], r]
|
||
pos = e
|
||
parts.append(text[pos:b])
|
||
return "".join(parts)
|