"""
plate_reader.py — reads Saudi license plates with EasyOCR (Arabic + English).
v3 (2026-09) — what changed and why
-----------------------------------
1. Plate LOCALIZATION first (OpenCV): bright, text-bearing rectangles with a
plate-like aspect ratio are found and perspective-warped flat BEFORE any
OCR. The old version anchored the crop on "the most confident digit in the
whole photo", which could land on a watermark or random text and then read
garbage (e.g. ":7.7 ES"). The old digit-anchor crop is still tried as a
fallback candidate.
2. CROSS-CHECK instead of translate-after: a Saudi plate says everything twice
(Arabic-Indic digits over Western digits, Arabic letters over English
letters). Each character POSITION collects votes from BOTH rows and from
several OCR passes (free text + English allowlist + Arabic allowlist), then
the two rows are fused position by position through the official
17-letter table. Agreement = confirmed; disagreement = best-voted char,
flagged for review.
3. Orientation is tested, not assumed: the Arabic row is aligned against the
English row both directly and reversed (OCR engines disagree on Arabic
string order); the alignment with the most agreement wins.
4. Every candidate crop is scored (agreement + plate-format validity) and the
best one is returned; the search stops early on a fully confirmed read.
Output shape is unchanged (available / raw_text_blocks / fields / readings),
with extra keys: status ("confirmed" | "review"), status_ar, agreement,
positions (per-character detail), source (which crop won).
plate_reader_paddle.py reuses run_pipeline() from here with PaddleOCR as the
engine, so A/B differences come from the OCR engine only.
"""
import os
import re
import sys
import numpy as np
from PIL import Image, ImageOps
CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".model_cache", "easyocr")
_reader = None
_load_lock = __import__("threading").Lock()
def _log(msg):
print(f"[plate_reader] {msg}", file=sys.stderr, flush=True)
def _gpu():
try:
import torch
return torch.cuda.is_available()
except Exception:
return False
def _get_reader():
"""Loads EasyOCR once. Thread-safe (warm-up and a request may call this
at the same time), and a failed load is retried on the next call
instead of being remembered forever."""
global _reader
if _reader is not None:
return _reader
with _load_lock:
if _reader is not None:
return _reader
try:
import easyocr
_log("Loading EasyOCR (Arabic + English, may download on first run)...")
os.makedirs(CACHE_DIR, exist_ok=True)
_reader = easyocr.Reader(["ar", "en"], model_storage_directory=CACHE_DIR,
download_enabled=True, gpu=_gpu())
_log("Loaded.")
except Exception as e:
import traceback
_log(f"Failed to load: {e}\n{traceback.format_exc()}")
_reader = None
return _reader
# ---------------------------------------------------------------------------
# Official Saudi plate alphabet (17 letters). Kept under the old names because
# other modules import them.
# ---------------------------------------------------------------------------
_AR_TO_EN_LETTER = {
"أ": "A", "ب": "B", "ح": "J", "د": "D", "ر": "R", "س": "S", "ص": "X",
"ط": "T", "ع": "E", "ق": "G", "ك": "K", "ل": "L", "م": "Z", "ن": "N",
"هـ": "H", "ه": "H", "و": "U", "ى": "V",
}
_EN_TO_AR_LETTER = {v: k for k, v in _AR_TO_EN_LETTER.items() if k != "ه"}
_VALID_ARABIC_LETTERS = set(_AR_TO_EN_LETTER.keys()) | {"ه"}
_VALID_ENGLISH_LETTERS = set(_EN_TO_AR_LETTER.keys())
# Common OCR variants of the plate letters -> canonical plate letter.
_AR_VARIANTS = {"ا": "أ", "إ": "أ", "آ": "أ", "ي": "ى", "ة": "ه", "ھ": "ه", "ہ": "ه"}
_EASTERN_DIGITS = "٠١٢٣٤٥٦٧٨٩"
_PERSIAN_DIGITS = "۰۱۲۳۴۵۶۷۸۹"
_EASTERN_TO_WESTERN_DIGITS = str.maketrans(_EASTERN_DIGITS + _PERSIAN_DIGITS, "0123456789" * 2)
_ARABIC_ALLOWLIST = "".join(sorted(k for k in _VALID_ARABIC_LETTERS if len(k) == 1)) + _EASTERN_DIGITS
_ENGLISH_ALLOWLIST = "".join(sorted(_VALID_ENGLISH_LETTERS)) + "0123456789"
_MIN_CHAR_CONFIDENCE = 0.05
_BRANDING_TEXT_MARKERS = ["السعودية", "سعودية", "عربية", "المملكة", "KSA", "SAUDI", "ARABIA"]
_DIGIT_CHAR = re.compile(r"^[0-9\u0660-\u0669\u06F0-\u06F9]$")
_ARABIC_CHAR = re.compile(r"^[\u0600-\u06FF]$")
_LATIN_CHAR = re.compile(r"^[A-Za-z]$")
MAX_DIGITS = 4
MAX_LETTERS = 3
def _looks_like_branding(text):
stripped = text.strip()
if not stripped:
return False
upper = stripped.upper()
return any(m in stripped or m in upper for m in _BRANDING_TEXT_MARKERS)
def _classify_script(text):
for ch in text:
if _DIGIT_CHAR.match(ch):
return "digit"
if _ARABIC_CHAR.match(ch):
return "ar"
if _LATIN_CHAR.match(ch):
return "en"
return None
def _bbox_center_and_size(bbox):
xs = [p[0] for p in bbox]
ys = [p[1] for p in bbox]
return {"cx": sum(xs) / len(xs), "cy": sum(ys) / len(ys),
"w": max(xs) - min(xs), "h": max(ys) - min(ys)}
# Legacy helpers — no longer used by the pipeline, kept for import compatibility.
def _split_by_largest_gap(items, key):
if len(items) <= 1:
return [items]
s = sorted(items, key=key)
gaps = sorted(((key(s[i + 1]) - key(s[i]), i) for i in range(len(s) - 1)), reverse=True)
big, i = gaps[0]
others = [g for g, _ in gaps[1:]]
typical = sum(others) / len(others) if others else 0
return [s] if big < typical * 1.8 else [s[:i + 1], s[i + 1:]]
def _assemble_group(group):
if not group:
return None, "", 0.0
scripts = [d["script"] for d in group]
maj = max(set(scripts), key=scripts.count)
same = [d for d in group if d["script"] == maj]
ordered = sorted(same, key=(lambda d: -d["cx"]) if maj == "ar" else (lambda d: d["cx"]))
return maj, "".join(d["text"] for d in ordered), min(d["conf"] for d in same)
# ---------------------------------------------------------------------------
# 1. Plate localization
# ---------------------------------------------------------------------------
def _order_quad(pts):
pts = np.asarray(pts, dtype="float32")
s = pts.sum(1)
d = (pts[:, 1] - pts[:, 0])
return np.array([pts[np.argmin(s)], pts[np.argmin(d)], pts[np.argmax(s)], pts[np.argmax(d)]],
dtype="float32") # tl, tr, br, bl
def _overlap(a, b):
"""Intersection over the SMALLER box — treats a box nested inside
another as a duplicate."""
ix = max(0, min(a[2], b[2]) - max(a[0], b[0]))
iy = max(0, min(a[3], b[3]) - max(a[1], b[1]))
small = min((a[2] - a[0]) * (a[3] - a[1]), (b[2] - b[0]) * (b[3] - b[1]))
return ix * iy / small if small > 0 else 0
def _find_plate_regions(pil_image, max_n=2, inside_box=False):
"""Returns up to max_n (PIL crop, description) of likely plate rectangles,
perspective-corrected. Pure OpenCV, no model needed."""
try:
import cv2
except Exception:
return []
rgb = np.array(pil_image.convert("RGB"))
H, W = rgb.shape[:2]
s = min(1.0, 1000.0 / max(H, W))
small = cv2.resize(rgb, (int(W * s), int(H * s)), interpolation=cv2.INTER_AREA) if s < 1 else rgb
gray = cv2.bilateralFilter(cv2.cvtColor(small, cv2.COLOR_RGB2GRAY), 9, 60, 60)
h, w = gray.shape
edges = cv2.Canny(gray, 40, 140)
_, bright = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
local = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_MEAN_C, cv2.THRESH_BINARY, 51, -8)
masks = [
cv2.dilate(edges, np.ones((3, 3), np.uint8), iterations=1),
cv2.morphologyEx(bright, cv2.MORPH_OPEN, np.ones((3, 3), np.uint8)),
cv2.morphologyEx(local, cv2.MORPH_CLOSE, np.ones((5, 9), np.uint8)),
]
cands = []
for m in masks:
cnts, _ = cv2.findContours(m, cv2.RETR_LIST, cv2.CHAIN_APPROX_SIMPLE)
for c in cnts:
area = cv2.contourArea(c)
if area < 0.002 * w * h or area > 0.80 * w * h:
continue
rect = cv2.minAreaRect(c)
rw, rh = rect[1]
if min(rw, rh) < 12:
continue
aspect = max(rw, rh) / min(rw, rh)
if not 1.8 <= aspect <= 6.5:
continue
rectangularity = area / (rw * rh + 1e-6)
if rectangularity < 0.6:
continue
# Correct Perspective: use the plate's REAL 4 corners when the
# outline simplifies to a quadrilateral (a plate shot at an angle
# is a trapezoid, not a rotated rectangle); otherwise fall back
# to the rotated bounding rectangle.
quad = None
hull = cv2.convexHull(c)
peri = cv2.arcLength(hull, True)
for eps in (0.02, 0.03, 0.045):
approx = cv2.approxPolyDP(hull, eps * peri, True)
if len(approx) == 4:
q = _order_quad(approx.reshape(4, 2))
if cv2.contourArea(q) > 0.75 * area:
quad = q
break
if quad is None:
quad = _order_quad(cv2.boxPoints(rect))
top = quad[1] - quad[0]
side = quad[3] - quad[0]
if np.linalg.norm(top) <= np.linalg.norm(side):
continue # long side is vertical -> not a horizontal plate
if abs(np.degrees(np.arctan2(top[1], top[0]))) > 35:
continue
x, y, bw, bh = cv2.boundingRect(c)
brightness = gray[y:y + bh, x:x + bw].mean() / 255.0
edge_density = edges[y:y + bh, x:x + bw].mean() / 255.0
if edge_density < 0.04 or brightness < 0.35:
continue
# Bigger wins: tiny stickers/watermarks lose, and the WHOLE plate
# beats the individual white cells it is divided into.
size_factor = min(3.0, ((area / (w * h)) / 0.03) ** 0.5)
score = rectangularity * (0.5 + brightness) * min(edge_density * 5, 1.5) * size_factor
cands.append((score, quad / s, (x / s, y / s, (x + bw) / s, (y + bh) / s)))
cands.sort(key=lambda c: -c[0])
kept = []
for c in cands:
if all(_overlap(c[2], k[2]) < 0.6 for k in kept):
kept.append(c)
if len(kept) >= max_n:
break
out = []
for score, quad, box in kept:
center = quad.mean(0)
quad = (quad - center) * np.array([1.06, 1.15], dtype="float32") + center # small margin
quad[:, 0] = quad[:, 0].clip(0, W - 1)
quad[:, 1] = quad[:, 1].clip(0, H - 1)
dw = int(max(np.linalg.norm(quad[1] - quad[0]), np.linalg.norm(quad[2] - quad[3])))
dh = int(max(np.linalg.norm(quad[3] - quad[0]), np.linalg.norm(quad[2] - quad[1])))
if dw < 20 or dh < 10:
continue
if inside_box: # searching inside a user-drawn box: the plate fills much of it
if not (1.3 <= dw / dh <= 7.0 and dw * dh >= 0.2 * W * H):
continue
elif not _plate_shaped(dw, dh, W, H):
continue
dst = np.array([[0, 0], [dw - 1, 0], [dw - 1, dh - 1], [0, dh - 1]], dtype="float32")
warped = cv2.warpPerspective(rgb, cv2.getPerspectiveTransform(quad, dst), (dw, dh))
crop = Image.fromarray(warped)
crop.info["ts_quad"] = [float(v) for v in (quad / np.array([W, H], dtype="float32")).reshape(-1)]
out.append((crop, f"contour score={score:.2f} box={tuple(int(v) for v in box)}"))
return out
def _digit_anchor_crop(pil_image, first_pass):
"""Old heuristic, kept as a fallback candidate: crop around the most
confident digit detection in the full image."""
digit_dets = [(bbox, conf) for (bbox, text, conf) in first_pass
if conf >= 0.3 and any(_DIGIT_CHAR.match(ch) for ch in text)]
if not digit_dets:
return None
best_bbox, _ = max(digit_dets, key=lambda d: d[1])
pos = _bbox_center_and_size(best_bbox)
W, H = pil_image.size
pad_x = min(pos["w"] * 3.0, W * 0.18)
pad_y = min(pos["h"] * 4.0, H * 0.18)
box = (max(0, pos["cx"] - pos["w"] / 2 - pad_x), max(0, pos["cy"] - pos["h"] / 2 - pad_y),
min(W, pos["cx"] + pos["w"] / 2 + pad_x), min(H, pos["cy"] + pos["h"] / 2 + pad_y))
if box[2] - box[0] < 20 or box[3] - box[1] < 20:
return None
W, H = pil_image.size
crop = pil_image.crop(box)
x0, y0, x1, y1 = box
crop.info["ts_quad"] = [x0 / W, y0 / H, x1 / W, y0 / H, x1 / W, y1 / H, x0 / W, y1 / H]
return crop
def _prep(crop):
"""Upscale small crops so each character has enough pixels."""
w, h = crop.size
scale = min(4.0, 900.0 / w) if w < 900 else 900.0 / w
if abs(scale - 1.0) > 0.01:
crop = crop.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
return np.array(crop.convert("RGB"))
# ---------------------------------------------------------------------------
# 2. Decode OCR boxes into per-position votes, then fuse the two rows
# ---------------------------------------------------------------------------
def _norm_ar(ch):
return _AR_VARIANTS.get(ch, ch)
def _char_dets(results, ar_rtl=True):
"""Splits every OCR box into per-character detections with an estimated
x position. Letters are stored in ENGLISH space (Arabic mapped through
the official table) so both rows can be compared directly."""
blocks, dets = [], []
for bbox, text, conf in results:
conf = float(conf)
t = (text or "").strip()
blocks.append({"text": t, "confidence": round(conf, 3)})
if conf < _MIN_CHAR_CONFIDENCE or not t or _looks_like_branding(t):
continue
t = t.replace("ـ", "").replace(" ", "")
if not t:
continue
pos = _bbox_center_and_size(bbox)
n = len(t)
cw = pos["w"] / n if n else pos["w"]
left = pos["cx"] - pos["w"] / 2
right = pos["cx"] + pos["w"] / 2
arabic_box = any(_ARABIC_CHAR.match(c) and not _DIGIT_CHAR.match(c) for c in t)
for i, ch in enumerate(t):
# Arabic text runs right-to-left inside its box; digits/Latin left-to-right.
rtl = arabic_box and ar_rtl and not _DIGIT_CHAR.match(ch)
cx = right - (i + 0.5) * cw if rtl else left + (i + 0.5) * cw
base = {"cx": cx, "cy": pos["cy"], "h": pos["h"], "w": cw, "conf": conf}
if _DIGIT_CHAR.match(ch):
dets.append({**base, "kind": "digit", "eastern": ch in _EASTERN_DIGITS + _PERSIAN_DIGITS,
"char": ch.translate(_EASTERN_TO_WESTERN_DIGITS)})
elif _ARABIC_CHAR.match(ch):
a = _norm_ar(ch)
if a in _AR_TO_EN_LETTER:
dets.append({**base, "kind": "ar", "char": _AR_TO_EN_LETTER[a]})
elif ch.upper() in _VALID_ENGLISH_LETTERS:
dets.append({**base, "kind": "en", "char": ch.upper()})
return blocks, dets
def _normv(v):
t = sum(v.values())
return {k: x / t for k, x in v.items()} if t > 0 else {}
def _combine(vt, vb, raw):
"""One character position seen by the top row (vt) and/or the bottom
row (vb), both as {char: summed confidence} in the same char space.
Reported confidence = vote share of the winner x its mean OCR confidence."""
pt, pb = _normv(vt), _normv(vb)
if pt and pb:
comb = {}
for p in (pt, pb):
for k, x in p.items():
comb[k] = comb.get(k, 0) + x
ranked = sorted(comb.items(), key=lambda kv: -kv[1])
agreed = max(pt, key=pt.get) == max(pb, key=pb.get) == ranked[0][0]
return {"char": ranked[0][0], "confidence": round(ranked[0][1] / 2 * raw(ranked[0][0]), 3), "agreed": agreed,
"sources": ["top", "bottom"], "alternatives": [k for k, _ in ranked[1:3]]}
p, src = (pt, "top") if pt else (pb, "bottom")
ranked = sorted(p.items(), key=lambda kv: -kv[1])
return {"char": ranked[0][0], "confidence": round(ranked[0][1] * 0.6 * raw(ranked[0][0]), 3), "agreed": False,
"sources": [src], "alternatives": [k for k, _ in ranked[1:3]]}
def _joint_positions(top, bottom):
"""Clusters top-row and bottom-row detections TOGETHER by x position.
On a Saudi plate each Arabic character sits directly above its English
twin, so characters that share an x position are the same character
read twice — that is the cross-check."""
items = [(d, "top") for d in top] + [(d, "bottom") for d in bottom]
if not items:
return []
cw = float(np.median([d["w"] for d, _ in items])) or 1.0
clusters = []
for d, src in sorted(items, key=lambda it: it[0]["cx"]):
if clusters and d["cx"] - clusters[-1]["cx"] < 0.55 * cw:
c = clusters[-1]
c["n"] += 1
c["cx"] += (d["cx"] - c["cx"]) / c["n"]
else:
c = {"cx": d["cx"], "n": 1, "top": {}, "bottom": {}, "sum": {}, "cnt": {}}
clusters.append(c)
c[src][d["char"]] = c[src].get(d["char"], 0) + d["conf"]
c["sum"][d["char"]] = c["sum"].get(d["char"], 0) + d["conf"]
c["cnt"][d["char"]] = c["cnt"].get(d["char"], 0) + 1
mass = [sum(c["top"].values()) + sum(c["bottom"].values()) for c in clusters]
top_mass = max(mass)
kept = [c for c, m in zip(clusters, mass) if m >= 0.15 * top_mass]
positions = [dict(_combine(c["top"], c["bottom"], lambda ch, c=c: min(1.0, c["sum"][ch] / c["cnt"][ch])),
cx=c["cx"]) for c in kept]
# Characters of one field sit next to each other. A cluster far away
# (e.g. the emblem/picture in the middle of the plate read as "88") is
# a separate group — keep only the strongest group.
groups, cur = [], []
for pos in positions:
if cur and pos["cx"] - cur[-1]["cx"] > 2.2 * cw:
groups.append(cur)
cur = []
cur.append(pos)
if cur:
groups.append(cur)
if len(groups) > 1:
def strength(g):
return (2 * sum(p["agreed"] for p in g) + len(g) + sum(p["confidence"] for p in g),
-g[0]["cx"])
positions = max(groups, key=strength)
return positions
def _seq_fusion(top, bottom, cap):
"""Sequence-level cross-check. Each row is read on its own (left->right),
then the rows are aligned by CONTENT, allowing the longer row to contain
extra junk characters. Fixes cases where x-positions mislead — e.g. the
Arabic row read as 4 letters ("أدطى") inside one box while English read
3 ("VTD"): dropping the extra letter lines them up perfectly."""
from itertools import combinations
a = _joint_positions(top, [])
b = _joint_positions([], bottom)
if not a or not b:
return None
short, long_, short_is_a = (a, b, True) if len(a) <= len(b) else (b, a, False)
if len(long_) > 7:
return None
best = None
for idx in combinations(range(len(long_)), len(short)):
pairs = [(short[i], long_[j]) for i, j in enumerate(idx)]
matches = sum(p["char"] == q["char"] for p, q in pairs)
key = (matches, sum(p["confidence"] + q["confidence"] for p, q in pairs))
if best is None or key > best[0]:
best = (key, pairs)
fused = []
for p, q in best[1]:
pa, pb = (p, q) if short_is_a else (q, p)
if pa["char"] == pb["char"]:
fused.append({"char": pa["char"], "confidence": round(max(pa["confidence"], pb["confidence"]) / 0.6, 3),
"agreed": True, "sources": ["top", "bottom"],
"alternatives": pa["alternatives"] or pb["alternatives"]})
else:
win, lose = (pa, pb) if pa["confidence"] >= pb["confidence"] else (pb, pa)
fused.append({"char": win["char"], "confidence": round(win["confidence"] * 0.8, 3),
"agreed": False, "sources": ["top", "bottom"],
"alternatives": [lose["char"]] + [c for c in win["alternatives"] if c != lose["char"]][:1]})
for f in fused:
f["confidence"] = min(f["confidence"], 1.0)
return _trim(fused, cap)
def _best_fusion(top, bottom, cap):
"""Position-based fusion vs sequence-based fusion — keep whichever makes
the two rows agree more."""
by_x = _trim(_joint_positions(top, bottom), cap)
by_seq = _seq_fusion(top, bottom, cap)
if not by_seq:
return by_x
def quality(ps):
return (sum(p["agreed"] for p in ps), -abs(len(ps) - cap) if cap == MAX_LETTERS else 0,
sum(p["confidence"] for p in ps))
return by_seq if quality(by_seq) > quality(by_x) else by_x
def _trim(positions, cap):
if len(positions) <= cap:
return positions
keep = sorted(range(len(positions)), key=lambda i: -positions[i]["confidence"])[:cap]
return [positions[i] for i in sorted(keep)]
def _spaced(text):
"""Plates show each letter separately: "ب ب ب" / "B B B". (Arabic
letters written together would join into one word.)"""
if not text:
return text
out = []
for ch in text: # keep a letter and its tatweel together: "هـ", not "ه ـ"
if ch == "\u0640" and out:
out[-1] += ch
else:
out.append(ch)
return " ".join(out)
def _en_to_ar_logical(en_text):
"""English row (left->right) -> Arabic string in reading order.
On Saudi plates each Arabic letter sits directly above its English
counterpart, so the Arabic reading order (right->left) is the English
order reversed."""
return "".join(_EN_TO_AR_LETTER.get(c, "") for c in reversed(en_text))
def _decode(results):
"""Tries both placements for multi-letter Arabic boxes (engines differ on
whether they return Arabic in reading or visual order) and keeps the
placement whose letters agree best with the English row."""
a = _decode_once(results, ar_rtl=True)
b = _decode_once(results, ar_rtl=False)
if a is None or b is None:
return a or b
return a if a["score"] >= b["score"] else b
def _decode_once(results, ar_rtl):
"""Turns raw OCR output for ONE plate crop into a scored reading."""
blocks, dets = _char_dets(results, ar_rtl)
if not dets:
return None
med_h = float(np.median([d["h"] for d in dets]))
dets = [d for d in dets if d["h"] >= 0.5 * med_h] # drops small "KSA"/emblem text
ys = [d["cy"] for d in dets]
two_rows = (max(ys) - min(ys)) > 0.6 * med_h
y_mid = (max(ys) + min(ys)) / 2
def row(d):
return ("top" if d["cy"] < y_mid else "bottom") if two_rows else None
top_digits, bot_digits, ar_letters, en_letters = [], [], [], []
for d in dets:
r = row(d)
if d["kind"] == "digit":
if r == "top" or (r is None and d["eastern"]):
top_digits.append(d)
else:
bot_digits.append(d)
elif d["kind"] == "ar" and r in ("top", None):
ar_letters.append(d)
elif d["kind"] == "en" and r in ("bottom", None):
en_letters.append(d)
digits = _best_fusion(top_digits, bot_digits, MAX_DIGITS)
letters = _best_fusion(ar_letters, en_letters, MAX_LETTERS)
orient = "arabic-rtl" if ar_rtl else "arabic-visual"
if not digits and not letters:
return None
d_text = "".join(p["char"] for p in digits)
l_text = "".join(p["char"] for p in letters)
allp = digits + letters
n_agreed = sum(p["agreed"] for p in allp)
mean_conf = sum(p["confidence"] for p in allp) / len(allp)
format_ok = 1 <= len(d_text) <= MAX_DIGITS and 1 <= len(l_text) <= MAX_LETTERS
confirmed = format_ok and len(l_text) == MAX_LETTERS and n_agreed == len(allp)
score = 2 * n_agreed + len(allp) + mean_conf + (3 if format_ok else 0) + (1 if len(l_text) == 3 else 0)
derived_side = None
if letters and not ar_letters:
derived_side = "ar"
elif letters and not en_letters:
derived_side = "en"
return {
"blocks": blocks, "digits": digits, "letters": letters, "digits_text": d_text,
"letters_en": l_text, "letters_ar": _en_to_ar_logical(l_text), "orientation": orient,
"n_agreed": n_agreed, "n_total": len(allp), "mean_conf": mean_conf,
"format_ok": format_ok, "confirmed": confirmed, "score": score, "derived_side": derived_side,
}
def _alternate_readings(best):
"""Best reading + up to 2 alternates made by swapping the weakest
uncertain position to its runner-up character."""
base_d = [p["char"] for p in best["digits"]]
base_l = [p["char"] for p in best["letters"]]
def pack(d, l, conf):
en = "".join(l)
return {"letters_ar": _spaced(_en_to_ar_logical(en)), "letters_en": _spaced(en), "digits": "".join(d),
"confidence": round(conf, 3)}
readings = [pack(base_d, base_l, best["mean_conf"])]
weak = [("d", i, p) for i, p in enumerate(best["digits"]) if not p["agreed"] and p["alternatives"]]
weak += [("l", i, p) for i, p in enumerate(best["letters"]) if not p["agreed"] and p["alternatives"]]
weak.sort(key=lambda t: t[2]["confidence"])
for kind, i, p in weak[:2]:
d, l = list(base_d), list(base_l)
(d if kind == "d" else l)[i] = p["alternatives"][0]
readings.append(pack(d, l, best["mean_conf"] * 0.8))
return readings
def _empty(blocks):
return {"available": True, "status": "review",
"status_ar": "تعذّرت قراءة اللوحة — يُرجى التحديد يدويًا",
"fields": {"digits": None, "letters_ar": None, "letters_en": None, "derived_side": None},
"readings": []}
# ---------------------------------------------------------------------------
# 3. Engine-agnostic pipeline
# ---------------------------------------------------------------------------
def _good_enough(r):
return r is not None and (r["confirmed"] or (r["format_ok"] and len(r["letters_en"]) == MAX_LETTERS
and r["n_agreed"] >= r["n_total"] - 1))
def _manual_crop(pil_image, box):
"""box = (x0, y0, x1, y1) as fractions 0..1 of the image, drawn by the
user — or 8 numbers = the plate's 4 corners clicked by the user, which
are perspective-corrected (straightened) before reading."""
W, H = pil_image.size
if len(box) == 8:
import image_enhancer
pts = np.asarray(box, dtype="float32").reshape(4, 2) * np.array([W, H], dtype="float32")
crop = Image.fromarray(image_enhancer.warp_quad(np.asarray(pil_image.convert("RGB")), pts, margin=0.04))
crop.info["ts_quad"] = [float(v) for v in box]
return crop
x0, y0, x1, y1 = [min(1.0, max(0.0, float(v))) for v in box]
x0, x1 = sorted((x0, x1))
y0, y1 = sorted((y0, y1))
px, py = (x1 - x0) * 0.04, (y1 - y0) * 0.08
bx0, by0 = int(max(0, x0 - px) * W), int(max(0, y0 - py) * H)
bx1, by1 = int(min(1, x1 + px) * W), int(min(1, y1 + py) * H)
crop = pil_image.crop((bx0, by0, bx1, by1))
crop.info["ts_quad"] = [x0, y0, x1, y0, x1, y1, x0, y1]
# The user only draws a rough rectangle: find the plate's own corners
# inside it and straighten a tilted plate automatically.
try:
found = _find_plate_regions(crop, max_n=1, inside_box=True)
if found:
warped, desc = found[0]
q = warped.info.get("ts_quad")
cw, ch = crop.size
warped.info["ts_quad"] = [(bx0 + q[i] * cw) / W if i % 2 == 0 else (by0 + q[i] * ch) / H for i in range(8)]
_log(f"manual box refined to the plate corners ({desc})")
return warped
except Exception as e:
_log(f"corner refinement failed: {e}")
return crop
# ---------------------------------------------------------------------------
# Dark / low-contrast enhancement — the same chain forensic tools such as
# Amped FIVE use for night plates: Exposure -> Curves -> Extract Channel ->
# Levels -> Sharpening. (Crop / Correct Perspective / Resize are already done
# by the localization + _prep steps. Frame Averaging needs several video
# frames, so it doesn't apply to a single photo.)
# ---------------------------------------------------------------------------
def _brightness(pil_image):
g = np.asarray(pil_image.convert("L"), dtype=np.float32) / 255.0
return float(np.median(g)), float(g.std())
def _needs_enhance(pil_image):
med, std = _brightness(pil_image)
return med < 0.38 or std < 0.16
def _enhance(pil_image):
import cv2
rgb = np.asarray(pil_image.convert("RGB"), dtype=np.float32) / 255.0
# 1) Exposure: gamma so the median brightness lands near mid-grey.
med = float(np.median(rgb.mean(2)))
if 0.02 < med < 0.45:
gamma = np.log(0.5) / np.log(med)
rgb = np.power(rgb, max(0.35, min(1.0, gamma)))
u8 = (np.clip(rgb, 0, 1) * 255).astype(np.uint8)
# 2) Curves: local contrast (CLAHE) on lightness only.
lab = cv2.cvtColor(u8, cv2.COLOR_RGB2LAB)
lab[..., 0] = cv2.createCLAHE(clipLimit=2.5, tileGridSize=(8, 8)).apply(lab[..., 0])
u8 = cv2.cvtColor(lab, cv2.COLOR_LAB2RGB)
# 3) Extract channel: the channel where the plate text contrasts most
# (tinted night light often kills one channel and leaves another clean).
chans = [u8[..., 0], u8[..., 1], u8[..., 2], cv2.cvtColor(u8, cv2.COLOR_RGB2GRAY)]
ch = max(chans, key=lambda c: float(c.std()))
# 4) Levels: stretch the 1st..99th percentile to full black..white.
lo, hi = np.percentile(ch, (1, 99))
if hi - lo > 10:
ch = np.clip((ch.astype(np.float32) - lo) * 255.0 / (hi - lo), 0, 255).astype(np.uint8)
# 5) Sharpening (unsharp mask), after a light denoise so sensor noise in
# dark shots isn't amplified.
ch = cv2.fastNlMeansDenoising(ch, None, h=7, templateWindowSize=7, searchWindowSize=15)
blur = cv2.GaussianBlur(ch, (0, 0), 1.2)
ch = cv2.addWeighted(ch, 1.6, blur, -0.6, 0)
return Image.fromarray(cv2.cvtColor(ch, cv2.COLOR_GRAY2RGB))
def enhance_for_view(pil_image):
"""Public helper (same chain) — e.g. to show the user the brightened image."""
return _enhance(ImageOps.exif_transpose(pil_image).convert("RGB"))
def run_pipeline(pil_image, engines, ocr_full, tag="plate", manual_box=None):
"""engines = [(name, fn)], fn(np_rgb) -> [(bbox, text, conf)] for a plate
crop; every engine reads every crop tried. ocr_full(np_rgb) -> same, for
the whole photo."""
import time
t_start = time.perf_counter()
pil_image = ImageOps.exif_transpose(pil_image).convert("RGB")
if manual_box:
candidates = [(_manual_crop(pil_image, manual_box), "manual selection")]
_log(f"[{tag}] using the plate area selected by the user: {manual_box}")
else:
candidates = []
try: # trained plate locator first (Roboflow saudi_ALPR), when the key is set
import plate_chars_roboflow
if plate_chars_roboflow.available():
W, H = pil_image.size
for x0, y0, x1, y1, c in plate_chars_roboflow.locate_plates(np.asarray(pil_image)):
if not _plate_shaped(x1 - x0, y1 - y0, W, H):
_log(f"[{tag}] locator box rejected (not plate-shaped): {int(x1 - x0)}x{int(y1 - y0)} in {W}x{H}")
continue
mx, my = (x1 - x0) * 0.06, (y1 - y0) * 0.12
crop = pil_image.crop((int(max(0, x0 - mx)), int(max(0, y0 - my)),
int(min(W, x1 + mx)), int(min(H, y1 + my))))
crop.info["ts_quad"] = [x0 / W, y0 / H, x1 / W, y0 / H, x1 / W, y1 / H, x0 / W, y1 / H]
candidates.append((crop, f"plate locator {c:.2f}"))
except Exception as e:
_log(f"[{tag}] plate locator failed: {e}")
candidates += _find_plate_regions(pil_image)
if not candidates and _needs_enhance(pil_image):
# Dark photo: the plate edges may only show up after brightening.
# The enhanced image is used only to FIND the plate; the crop is
# taken from the original so the reading step can choose.
boxes = _find_plate_regions(_enhance(pil_image))
candidates = [(c, d + " (found after enhancement)") for c, d in boxes]
_log(f"[{tag}] plate regions found: {[d for _, d in candidates]}")
found = bool(candidates) and not manual_box
best, all_blocks, pool = None, [], []
def consider(crop, source, quad=None):
nonlocal best
prepped = _prep(crop)
good = False
for name, fn in engines:
try:
results = fn(prepped)
except Exception as e:
_log(f"[{name}] OCR failed on {source}: {e}")
continue
r = _decode(results)
if r is None:
continue
r["source"] = source
r["quad"] = quad
r["crop_img"] = crop
all_blocks.extend(r["blocks"])
pool.append(r)
_log(f"[{name}] {source}: {r['digits_text']} {r['letters_en']} ({r['letters_ar']}) "
f"agreed={r['n_agreed']}/{r['n_total']} score={r['score']:.2f}")
if best is None or r["score"] > best["score"]:
best = r
good = good or _good_enough(r)
# Good enough to stop searching: full format and at most one
# position where the two rows disagree.
return good
done = False
for crop, desc in candidates:
quad = crop.info.get("ts_quad")
dark = _needs_enhance(crop)
# Every crop is also read after the automatic enhancement chain (the
# same filters as the image-processing page). Dark/flat crops are
# read enhanced FIRST; otherwise enhancement is a second try.
variants = ["auto", "clahe", "plain"] if dark else ["plain", "auto"]
if crop.height < SMALL_PLATE_PX: # small/far plate: enlarge it first
variants = ["sr"] + variants
for v in variants:
if v == "sr":
img, label = _superres(crop), desc + " +superres"
elif v == "plain":
img, label = crop, desc
elif v == "clahe":
img, label = _enhance(crop), desc + " +enhanced"
else:
img, label = _auto_enhance(crop), desc + " +auto"
if img is None:
continue
if consider(img, label, quad):
done = True
break
if done:
break
if not done and not manual_box and not (best and best["format_ok"]):
try:
first_pass = ocr_full(np.array(pil_image))
except Exception as e:
_log(f"[{tag}] full-image OCR failed: {e}")
first_pass = []
anchor = _digit_anchor_crop(pil_image, first_pass)
if anchor is not None:
done = consider(anchor, "digit-anchor fallback", anchor.info.get("ts_quad"))
if best is None and first_pass:
r = _decode(first_pass)
if r:
r["source"] = "full image"
best = r
pool.append(r)
all_blocks.extend(r["blocks"])
_log(f"[{tag}] plate read in {time.perf_counter() - t_start:.1f}s")
if best is None:
out = _empty(all_blocks)
out["needs_selection"] = not manual_box
out["found"] = found
return out
ranked = _rank_readings(pool)
full = [g for g in ranked if _full_format(g["result"]) and g["result"]["mean_conf"] >= MIN_OPTION_CONF]
top = (full or ranked or [{"result": best}])[0]["result"]
conf = round(top["mean_conf"], 3)
status = "confirmed" if top["confirmed"] else "review"
# Only complete readings (digits + 3 letters) are offered as options;
# partial guesses are not presented as answers.
readings = [_pack_reading(g) for g in full[:MAX_READINGS]]
if readings and len(readings) < MAX_READINGS: # second-choice characters of the top reading
seen = {(r["digits"], r["letters_en"]) for r in readings}
for alt in _alternate_readings(top)[1:]:
k = (alt["digits"], alt["letters_en"])
if k not in seen and len(readings) < MAX_READINGS and len(alt["letters_en"].replace(" ", "")) == MAX_LETTERS:
seen.add(k)
readings.append(alt)
return {
"available": True,
"status": status if readings else "review",
"fields": {
"digits": {"text": top["digits_text"], "confidence": conf} if top["digits_text"] else None,
"letters_ar": {"text": _spaced(top["letters_ar"]), "confidence": conf} if top["letters_en"] else None,
"letters_en": {"text": _spaced(top["letters_en"]), "confidence": conf} if top["letters_en"] else None,
"derived_side": top["derived_side"],
},
"found": found,
# Where the plate was found (shown even when no complete reading).
"box": top.get("quad"),
"crop": _crop_b64(top.get("crop_img")),
# Not sure -> the UI asks the user to mark the plate.
"needs_selection": (not manual_box) and (not readings or not _good_enough(top)),
"readings": readings,
}
def _plate_shaped(w, h, W, H):
"""A plate is a small, wide rectangle: rejects boxes around a whole car
or a large part of the photo, and tall/square shapes."""
if w <= 0 or h <= 0:
return False
ratio = w / h
return 1.3 <= ratio <= 7.0 and (w * h) / float(W * H) <= 0.12 and w / float(W) <= 0.6
MIN_OPTION_CONF = 0.35 # readings below this are not offered as options
def _full_format(r):
return 1 <= len(r["digits_text"]) <= MAX_DIGITS and len(r["letters_en"]) == MAX_LETTERS
SMALL_PLATE_PX = 90
def _superres(crop):
"""4x super-resolution, then the automatic enhancement chain."""
try:
import plate_superres
big = plate_superres.upscale(crop)
if big is None:
return None
return _auto_enhance(big) or big
except Exception as e:
_log(f"super-resolution failed: {e}")
return None
def _auto_enhance(crop):
try:
import image_enhancer
return image_enhancer.enhance(crop)[0]
except Exception as e:
_log(f"auto enhancement failed: {e}")
return None
def _crop_b64(img):
if img is None:
return None
import base64, io
img = img.convert("RGB")
if img.width < 480:
s = 480 / img.width
img = img.resize((480, max(1, int(img.height * s))), Image.LANCZOS)
buf = io.BytesIO()
img.save(buf, "JPEG", quality=88)
return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
MAX_READINGS = 3
def _rank_readings(pool):
"""Group identical readings from different engines/crops; a reading
several engines agree on ranks above a single high-scoring one."""
groups = {}
for r in pool:
if not (r["digits_text"] or r["letters_en"]):
continue
k = (r["digits_text"], r["letters_en"])
g = groups.setdefault(k, {"result": r, "votes": 0, "best": r["score"]})
g["votes"] += 1
if r["score"] > g["best"]:
g["best"], g["result"] = r["score"], r
for g in groups.values():
g["rank"] = g["best"] + 0.15 * (g["votes"] - 1) + (0.2 if g["result"]["format_ok"] else 0)
return sorted(groups.values(), key=lambda g: -g["rank"])
def _pack_reading(g):
r = g["result"]
c = min(0.99, r["mean_conf"] + 0.05 * (g["votes"] - 1))
return {"letters_ar": _spaced(r["letters_ar"]), "letters_en": _spaced(r["letters_en"]),
"digits": r["digits_text"], "confidence": round(c, 3)}
def read_plate(pil_image, manual_box=None):
reader = _get_reader()
engines = []
try:
import plate_chars_roboflow as rf
for i in range(rf.char_model_count()):
engines.append((f"c{i + 1}", lambda arr, i=i: rf.detect(arr, index=i)))
except Exception:
pass
try:
import plate_chars_yolo
if plate_chars_yolo.available():
engines.append(("local", plate_chars_yolo.detect))
except Exception:
pass
def ocr_crop_easy(arr):
# SPEED: text DETECTION (the slow CRAFT network) runs ONCE per crop;
# only the cheap RECOGNITION step is repeated per alphabet. The
# English pass reads the bottom row, the Arabic pass the top row;
# row filtering in _decode discards each pass's reading of the
# other row, so no unrestricted pass is needed.
try:
from easyocr.utils import reformat_input
img, grey = reformat_input(arr)
horiz, free = reader.detect(img, reformat=False)
horiz, free = horiz[0], free[0]
if not horiz and not free:
return []
out = []
for allow in (_ENGLISH_ALLOWLIST, _ARABIC_ALLOWLIST):
out += reader.recognize(grey, horiz, free, allowlist=allow, reformat=False, batch_size=8)
return out
except Exception as e:
_log(f"fast path failed ({e}); using readtext")
out = []
for allow in (_ENGLISH_ALLOWLIST, _ARABIC_ALLOWLIST):
out += reader.readtext(arr, allowlist=allow)
return out
if reader is not None:
engines.append(("ocr", ocr_crop_easy))
if not engines:
return {"available": False}
def ocr_full(arr):
if reader is None:
fn = engines[0][1]
return fn(arr)
h, w = arr.shape[:2]
s = 1280.0 / max(h, w)
if s < 1:
import cv2
arr = cv2.resize(arr, (int(w * s), int(h * s)), interpolation=cv2.INTER_AREA)
return [([[x / s, y / s] for x, y in b], t, c) for b, t, c in reader.readtext(arr)]
return reader.readtext(arr)
return run_pipeline(pil_image, engines, ocr_full, manual_box=manual_box)
def read_texts(pil_image, max_side=800):
"""Quick whole-photo text read (Latin letters only) — used to spot badge
wordmarks like JEEP / TOYOTA / LAND CRUISER for the make/model guess."""
reader = _get_reader()
if reader is None:
return []
img = ImageOps.exif_transpose(pil_image).convert("RGB")
w, h = img.size
s = min(1.0, max_side / max(w, h))
if s < 1:
img = img.resize((int(w * s), int(h * s)), Image.LANCZOS)
try:
return [(t, float(c)) for _, t, c in reader.readtext(
np.array(img), allowlist="ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz- ")]
except Exception as e:
_log(f"read_texts failed: {e}")
return []
"""
plate_reader.py — reads Saudi license plates with EasyOCR (Arabic + English).
v3 (2026-09) — what changed and why
-----------------------------------
1. Plate LOCALIZATION first (OpenCV): bright, text-bearing rectangles with a
plate-like aspect ratio are found and perspective-warped flat BEFORE any
OCR. The old version anchored the crop on "the most confident digit in the
whole photo", which could land on a watermark or random text and then read
garbage (e.g. ":7.7 ES"). The old digit-anchor crop is still tried as a
fallback candidate.
2. CROSS-CHECK instead of translate-after: a Saudi plate says everything twice
(Arabic-Indic digits over Western digits, Arabic letters over English
letters). Each character POSITION collects votes from BOTH rows and from
several OCR passes (free text + English allowlist + Arabic allowlist), then
the two rows are fused position by position through the official
17-letter table. Agreement = confirmed; disagreement = best-voted char,
flagged for review.
3. Orientation is tested, not assumed: the Arabic row is aligned against the
English row both directly and reversed (OCR engines disagree on Arabic
string order); the alignment with the most agreement wins.
4. Every candidate crop is scored (agreement + plate-format validity) and the
best one is returned; the search stops early on a fully confirmed read.
Output shape is unchanged (available / raw_text_blocks / fields / readings),
with extra keys: status ("confirmed" | "review"), status_ar, agreement,
positions (per-character detail), source (which crop won).
plate_reader_paddle.py reuses run_pipeline() from here with PaddleOCR as the
engine, so A/B differences come from the OCR engine only.
"""
import os
import re
import sys
import numpy as np
from PIL import Image, ImageOps
CACHE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".model_cache", "easyocr")
_reader = None
_load_lock = __import__("threading").Lock()
def _log(msg):
print(f"[plate_reader] {msg}", file=sys.stderr, flush=True)
def _gpu():
try:
import torch
return torch.cuda.is_available()
except Exception:
return False
def _get_reader():
"""Loads EasyOCR once. Thread-safe (warm-up and a request may call this
at the same time), and a failed load is retried on the next call
instead of being remembered forever."""
global _reader
if _reader is not None:
return _reader
with _load_lock:
if _reader is not None:
return _reader
try:
import easyocr
_log("Loading EasyOCR (Arabic + English, may download on first run)...")
os.makedirs(CACHE_DIR, exist_ok=True)
_reader = easyocr.Reader(["ar", "en"], model_storage_directory=CACHE_DIR,
download_enabled=True, gpu=_gpu())
_log("Loaded.")
except Exception as e:
import traceback
_log(f"Failed to load: {e}\n{traceback.format_exc()}")
_reader = None
return _reader
# ---------------------------------------------------------------------------
# Official Saudi plate alphabet (17 letters). Kept under the old names because
# other modules import them.
# ---------------------------------------------------------------------------
_AR_TO_EN_LETTER = {
"أ": "A", "ب": "B", "ح": "J", "د": "D", "ر": "R", "س": "S", "ص": "X",
"ط": "T", "ع": "E", "ق": "G", "ك": "K", "ل": "L", "م": "Z", "ن": "N",
"هـ": "H", "ه": "H", "و": "U", "ى": "V",
}
_EN_TO_AR_LETTER = {v: k for k, v in _AR_TO_EN_LETTER.items() if k != "ه"}
_VALID_ARABIC_LETTERS = set(_AR_TO_EN_LETTER.keys()) | {"ه"}
_VALID_ENGLISH_LETTERS = set(_EN_TO_AR_LETTER.keys())
# Common OCR variants of the plate letters -> canonical plate letter.
_AR_VARIANTS = {"ا": "أ", "إ": "أ", "آ": "أ", "ي": "ى", "ة": "ه", "ھ": "ه", "ہ": "ه"}
_EASTERN_DIGITS = "٠١٢٣٤٥٦٧٨٩"
_PERSIAN_DIGITS = "۰۱۲۳۴۵۶۷۸۹"
_EASTERN_TO_WESTERN_DIGITS = str.maketrans(_EASTERN_DIGITS + _PERSIAN_DIGITS, "0123456789" * 2)