Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions anpr_ocr/alpr.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,12 @@
from anpr_ocr.default_detector import DefaultDetector
from anpr_ocr.default_ocr import DefaultOCR
from anpr_ocr.logger import PlateLogger, VehicleRecord
from anpr_ocr.utils import PlateTracker, disambiguate_plate, pad_bounding_box
from anpr_ocr.utils import (
PlateTracker,
disambiguate_plate,
is_uae_region,
pad_bounding_box,
)


# pylint: disable=too-many-arguments, too-many-locals
Expand Down Expand Up @@ -306,7 +311,11 @@ def predict(self, frame: np.ndarray | str | os.PathLike) -> list[ALPRResult]:

# Apply syntax disambiguation and auto-healing
if ocr_result and ocr_result.text:
disambiguated_text = disambiguate_plate(ocr_result.text, self.syntax_pattern)
disambiguated_text = disambiguate_plate(
ocr_result.text,
self.syntax_pattern,
apply_indian_healing=not is_uae_region(ocr_result.region),
)
if disambiguated_text != ocr_result.text:
ocr_result = OcrResult(
text=disambiguated_text,
Expand Down Expand Up @@ -584,6 +593,7 @@ def draw_predictions_video(
smoothed_results: list[ALPRResult] = []

for box, display_text, display_conf, _ in tracked:
orig_res = det_map.get(id(box))
if logger is not None:
logger.observe(
plate_text=display_text,
Expand All @@ -592,8 +602,10 @@ def draw_predictions_video(
frame_idx=frame_idx,
frame_bgr=frame,
fps=fps,
model_region=(
orig_res.ocr.region if orig_res and orig_res.ocr else None
),
)
orig_res = det_map.get(id(box))
reg = orig_res.ocr.region if orig_res and orig_res.ocr else None
reg_c = (
orig_res.ocr.region_confidence if orig_res and orig_res.ocr else None
Expand Down
13 changes: 9 additions & 4 deletions anpr_ocr/default_ocr.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
heal_indian_plate,
is_two_row_plate,
split_two_row_crop,
is_uae_region,
)

# pylint: disable=too-many-arguments
Expand Down Expand Up @@ -137,10 +138,11 @@ def predict(self, cropped_plate: np.ndarray) -> OcrResult | None:

# 1. Baseline 1-row OCR prediction
t1, c1, r1, rp1 = self._run_single_crop(cropped_plate)
uae_plate = is_uae_region(r1)
if self.syntax_pattern and t1:
t1 = disambiguate_plate(t1, self.syntax_pattern)
t1 = disambiguate_plate(t1, self.syntax_pattern, apply_indian_healing=not uae_plate)

h1 = heal_indian_plate(t1)
h1 = t1 if uae_plate else heal_indian_plate(t1)
mean_c1 = (sum(c1) / len(c1)) if c1 else 0.0

# 2. Check for square / 2-row plate (e.g. auto-rickshaws, motorbikes)
Expand All @@ -154,9 +156,12 @@ def predict(self, cropped_plate: np.ndarray) -> OcrResult | None:
if len(tt) >= 2 and len(tb) >= 2:
t2 = tt + tb
c2 = ct + cb
uae_two_row_plate = is_uae_region(rt)
if self.syntax_pattern:
t2 = disambiguate_plate(t2, self.syntax_pattern)
h2 = heal_indian_plate(t2)
t2 = disambiguate_plate(
t2, self.syntax_pattern, apply_indian_healing=not uae_two_row_plate
)
h2 = t2 if uae_two_row_plate else heal_indian_plate(t2)
mean_c2 = (sum(c2) / len(c2)) if c2 else 0.0

# Prefer 2-row if:
Expand Down
5 changes: 3 additions & 2 deletions anpr_ocr/logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
import cv2
import numpy as np

from anpr_ocr.utils import get_state_name
from anpr_ocr.utils import get_plate_region


def is_similar_plate(p1: str, p2: str, threshold: float = 0.75) -> bool:
Expand Down Expand Up @@ -140,6 +140,7 @@ def observe(
frame_idx: int,
frame_bgr: np.ndarray,
fps: float = 25.0,
model_region: str | None = None,
) -> None:
"""
Record a plate observation from a video frame. If the vehicle is already
Expand All @@ -160,7 +161,7 @@ def observe(
x2, y2 = min(frame_bgr.shape[1], b.x2), min(frame_bgr.shape[0], b.y2)
crop = frame_bgr[y1:y2, x1:x2].copy() if (x2 > x1 and y2 > y1) else None

state = get_state_name(clean_text) or "Other / International"
state = get_plate_region(clean_text, model_region)
vehicle_color = estimate_vehicle_color(frame_bgr, b)

# Check if this detection matches an active vehicle event
Expand Down
38 changes: 35 additions & 3 deletions anpr_ocr/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,36 @@ def get_state_name(plate_text: str) -> str:
return INDIAN_STATE_NAMES.get(prefix, "")


UAE_REGION_ALIASES = frozenset(
{
"AE",
"UAE",
"UNITED ARAB EMIRATES",
"EMIRATES",
}
)


def is_uae_region(model_region: str | None) -> bool:
"""Return whether OCR region metadata identifies the United Arab Emirates."""
normalized_region = " ".join((model_region or "").upper().split())
return normalized_region in UAE_REGION_ALIASES or "UNITED ARAB EMIRATES" in normalized_region


def get_plate_region(plate_text: str, model_region: str | None = None) -> str:
"""Resolve an Indian state or UAE country from OCR text and model metadata."""
state_name = get_state_name(plate_text)
if state_name:
return state_name

normalized_region = " ".join((model_region or "").upper().split())
if is_uae_region(model_region):
return "United Arab Emirates"
if normalized_region:
return model_region or "Other / International"
return "Other / International"


STATE_PREFIX_CORRECTIONS: dict[str, str] = {
"0L": "DL",
"OL": "DL",
Expand Down Expand Up @@ -573,6 +603,7 @@ def disambiguate_plate(
pattern_mask: str | Sequence[str] | None = None,
custom_letter_to_digit: dict[str, str] | None = None,
custom_digit_to_letter: dict[str, str] | None = None,
apply_indian_healing: bool = True,
) -> str:
"""
Disambiguate visually similar characters (e.g. 0 vs O, 1 vs I, 8 vs B)
Expand All @@ -593,8 +624,9 @@ def disambiguate_plate(

clean_text = text.replace(" ", "").upper()

# Apply Indian plate healing first if eligible
clean_text = heal_indian_plate(clean_text)
# Apply Indian plate healing only when the OCR region is not UAE.
if apply_indian_healing:
clean_text = heal_indian_plate(clean_text)

if not pattern_mask:
return clean_text
Expand Down Expand Up @@ -642,4 +674,4 @@ def _score_candidate(cand: str) -> int:
result_chars.append(char)

disambiguated = "".join(result_chars)
return heal_indian_plate(disambiguated)
return heal_indian_plate(disambiguated) if apply_indian_healing else disambiguated
17 changes: 13 additions & 4 deletions anpr_ocr/video_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from anpr_ocr import ALPR, PlateLogger
from anpr_ocr.alpr import SUPPORTED_VIDEO_EXTS
from anpr_ocr.logger import estimate_vehicle_color
from anpr_ocr.utils import get_state_name
from anpr_ocr.utils import get_plate_region

# pylint: disable=too-many-branches, too-many-statements, import-outside-toplevel
# ruff: noqa: PLR0912, PLR0915, PLC0415, E501, ARG001
Expand All @@ -37,6 +37,11 @@
"LLDDLDDDD", # 9-char: TN45Q3566
"LLDDDDDD", # 8-char: LA020749
"LLDDDDD", # 7-char: IN03044
"LDDDDDD", # 7-char: UAE series + number
"LDDDDD", # 6-char: UAE series + number
"LLDDDD", # 6-char: UAE series + number
"DDDDDD", # 6-char: UAE numeric plate
"DDDDD", # 5-char: UAE numeric plate
"DLLDDDD", # 7-char: 5AU5341
"LLDDLLL", # 7-char: LB02APF
"LLLDDD", # 6-char: IZX842
Expand Down Expand Up @@ -243,6 +248,7 @@ def _play_video_live(
if frame_idx % frame_skip == 0:
res = alpr.predict(frame)
frame_dets = []
model_regions: dict[str, str | None] = {}
for r in res:
if not r.ocr or not r.ocr.text or len(r.ocr.text.strip()) < min_chars:
continue
Expand All @@ -253,29 +259,32 @@ def _play_video_live(
)
if conf < 0.35:
continue
frame_dets.append((r.detection.bounding_box, r.ocr.text.strip(), conf))
text = r.ocr.text.strip()
frame_dets.append((r.detection.bounding_box, text, conf))
model_regions[text] = r.ocr.region

tracked = tracker.update(frame_dets, frame_idx)
active_plates = [
(
box,
txt,
conf,
get_state_name(txt) or "Other / International",
get_plate_region(txt, model_regions.get(txt)),
estimate_vehicle_color(frame, box),
)
for box, txt, conf, _ in tracked
]

if logger is not None:
for box, txt, conf, _, _ in active_plates:
for box, txt, conf, state, _ in active_plates:
logger.observe(
plate_text=txt,
confidence=conf,
bounding_box=box,
frame_idx=frame_idx,
frame_bgr=frame,
fps=src_fps,
model_region=state,
)

# Render annotations
Expand Down
87 changes: 87 additions & 0 deletions test/test_alpr.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
"""

from pathlib import Path
from types import MethodType

import cv2
import numpy as np
Expand All @@ -23,7 +24,11 @@
split_two_row_crop,
vote_consensus_plate,
get_state_name,
get_plate_region,
is_uae_region,
)
from anpr_ocr.default_ocr import DefaultOCR
from anpr_ocr.video_demo import PLATE_PATTERNS

ASSETS_DIR = Path(__file__).resolve().parent.parent / "assets"

Expand Down Expand Up @@ -220,6 +225,88 @@ def test_indian_state_code_mapping() -> None:
assert get_state_name("XX01AB1234") == ""


@pytest.mark.parametrize(
"region",
["AE", "UAE", "United Arab Emirates", "emirates", " UAE "],
)
def test_uae_region_aliases(region: str) -> None:
assert is_uae_region(region) is True


@pytest.mark.parametrize("region", [None, "India", "United Kingdom", ""])
def test_non_uae_region_aliases(region: str | None) -> None:
assert is_uae_region(region) is False


def test_uae_plate_region_and_healing_guard() -> None:
assert get_plate_region("D123456", "UAE") == "United Arab Emirates"
assert get_plate_region("MH12DE1433", "UAE") == "Maharashtra"
assert get_plate_region("D123456", "United Kingdom") == "United Kingdom"
assert get_plate_region("D123456") == "Other / International"
assert disambiguate_plate("D123456", ["LDDDDDD"], apply_indian_healing=False) == "D123456"
assert disambiguate_plate("D123456", ["LDDDDDD"]) == "DL23456"


@pytest.mark.parametrize(
"plate, pattern",
[
("D123456", "LDDDDDD"),
("D12345", "LDDDDD"),
("AB1234", "LLDDDD"),
("123456", "DDDDDD"),
("12345", "DDDDD"),
],
)
def test_uae_syntax_patterns_preserve_plate_text(plate: str, pattern: str) -> None:
assert pattern in PLATE_PATTERNS
assert disambiguate_plate(plate, [pattern], apply_indian_healing=False) == plate


def test_default_ocr_preserves_uae_single_row_text() -> None:
ocr = DefaultOCR.__new__(DefaultOCR)
ocr.enhance_contrast = False
ocr.min_plate_width = 0
ocr.syntax_pattern = ["LDDDDDD"]

def run_single_crop(
_self: DefaultOCR, _crop: np.ndarray
) -> tuple[str, list[float], str | None, float | None]:
return "D123456", [0.95] * 7, "UAE", 0.99

ocr._run_single_crop = MethodType(run_single_crop, ocr) # type: ignore[method-assign]
result = ocr.predict(np.zeros((30, 100, 3), dtype=np.uint8))

assert result is not None
assert result.text == "D123456"
assert result.region == "UAE"


def test_default_ocr_preserves_uae_two_row_text() -> None:
ocr = DefaultOCR.__new__(DefaultOCR)
ocr.enhance_contrast = False
ocr.min_plate_width = 0
ocr.syntax_pattern = ["LDDDDD"]
calls = 0

def run_single_crop(
_self: DefaultOCR, _crop: np.ndarray
) -> tuple[str, list[float], str | None, float | None]:
nonlocal calls
calls += 1
if calls == 1:
return "D12", [0.60] * 3, "UAE", 0.99
if calls == 2:
return "D12", [0.95] * 3, "UAE", 0.99
return "345", [0.95] * 3, "UAE", 0.99

ocr._run_single_crop = MethodType(run_single_crop, ocr) # type: ignore[method-assign]
result = ocr.predict(np.zeros((100, 60, 3), dtype=np.uint8))

assert result is not None
assert result.text == "D12345"
assert result.region == "UAE"


def test_vehicle_color_estimation() -> None:
class DummyBox:
def __init__(self, x1: int, y1: int, x2: int, y2: int) -> None:
Expand Down
Loading