1import argparse
2import json
3from pathlib import Path
4
5import torch
6import torch.nn as nn
7from PIL import Image
8import torchvision.transforms as T
9
10HERE = Path(__file__).parent
11IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif", ".webp"}
12
13
14# ── Model Architecture ────────────────────────────────────────────────────────
15
16class ResBlock(nn.Module):
17 def __init__(self, in_ch, out_ch, stride=1):
18 super().__init__()
19 self.conv = nn.Sequential(
20 nn.Conv2d(in_ch, out_ch, 3, stride=stride, padding=1, bias=False),
21 nn.BatchNorm2d(out_ch),
22 nn.ReLU(inplace=True),
23 nn.Conv2d(out_ch, out_ch, 3, padding=1, bias=False),
24 nn.BatchNorm2d(out_ch),
25 )
26 self.downsample = None
27 if stride != 1 or in_ch != out_ch:
28 self.downsample = nn.Sequential(
29 nn.Conv2d(in_ch, out_ch, 1, stride=stride, bias=False),
30 nn.BatchNorm2d(out_ch),
31 )
32 self.relu = nn.ReLU(inplace=True)
33
34 def forward(self, x):
35 identity = x
36 out = self.conv(x)
37 if self.downsample is not None:
38 identity = self.downsample(x)
39 return self.relu(out + identity)
40
41
42class ResNetBiLSTMCTC(nn.Module):
43 def __init__(self, num_classes, hidden_size=256, num_layers=2, dropout=0.3):
44 super().__init__()
45 self.cnn = nn.Sequential(
46 nn.Conv2d(1, 64, 3, padding=1, bias=False),
47 nn.BatchNorm2d(64),
48 nn.ReLU(inplace=True),
49 nn.MaxPool2d(2, 2),
50 nn.Sequential(ResBlock(64, 64, stride=1)),
51 nn.Sequential(ResBlock(64, 128, stride=2)),
52 nn.Sequential(ResBlock(128, 256, stride=1)),
53 )
54 self.rnn = nn.LSTM(
55 input_size=256 * 16,
56 hidden_size=hidden_size,
57 num_layers=num_layers,
58 bidirectional=True,
59 dropout=dropout if num_layers > 1 else 0.0,
60 batch_first=False,
61 )
62 self.classifier = nn.Linear(hidden_size * 2, num_classes)
63
64 def forward(self, x):
65 feat = self.cnn(x) # [B, 256, H', W']
66 b, c, h, w = feat.shape
67 seq = feat.permute(3, 0, 1, 2) # [W', B, C, H']
68 seq = seq.reshape(w, b, c * h) # [W', B, C*H']
69 out, _ = self.rnn(seq) # [W', B, 2*hidden]
70 return self.classifier(out).log_softmax(2) # [W', B, num_classes]
71
72
73# ── Charset Parsing ───────────────────────────────────────────────────────────
74
75def load_charset(path):
76 """
77 Builds an index-to-character mapping lookup from char.json files.
78 Accommodates categories grouped by blocks, lists, or custom string token pairs.
79 """
80 with open(path, encoding="utf-8") as f:
81 data = json.load(f)
82
83 if isinstance(data, list):
84 return {i: ch for i, ch in enumerate(data)}
85
86 if isinstance(data, dict):
87 known_categories = {"khmer", "latin", "digits", "special"}
88 if known_categories & set(data.keys()):
89 order = ["khmer", "latin", "digits", "special"]
90 all_chars = "".join(data.get(cat, "") for cat in order)
91 return {0: "<blank>", **{i + 1: ch for i, ch in enumerate(all_chars)}}
92
93 try:
94 return {int(k): v for k, v in data.items()}
95 except ValueError:
96 pass
97
98 return {v: k for k, v in data.items()}
99
100 raise ValueError(f"Unrecognized token map schema inside character file: {path}")
101
102
103# ── Processing & Greedy Decoding ─────────────────────────────────────────────
104
105def ctc_decode(log_probs, idx2char, blank=0):
106 """Collapses consecutive duplicate indexes and strips blank tokens."""
107 indices = log_probs.argmax(dim=1).tolist()
108 chars, prev = [], None
109 for idx in indices:
110 if idx != prev and idx != blank:
111 chars.append(idx2char.get(idx, "?"))
112 prev = idx
113 return "".join(chars)
114
115
116_transform = T.Compose([
117 T.ToTensor(),
118 T.Normalize(mean=[0.5], std=[0.5]),
119])
120
121def preprocess(image_path, img_h=64, img_w=512):
122 """Converts target image asset to grayscale, scales to expected shape, and adds batch dims."""
123 img = Image.open(image_path).convert("L")
124 img = img.resize((img_w, img_h), Image.BICUBIC)
125 return _transform(img).unsqueeze(0)
126
127
128def predict_one(model, image_path, idx2char, device, blank=0):
129 tensor = preprocess(image_path).to(device)
130 with torch.no_grad():
131 log_probs = model(tensor)[:, 0, :] # Extracted time steps: [Time, Classes]
132 return ctc_decode(log_probs.cpu(), idx2char, blank=blank)
133
134
135# ── Core Runtime Entrypoint ───────────────────────────────────────────────────
136
137def main():
138 parser = argparse.ArgumentParser(description="Khmer OCR Inference System Stack")
139 parser.add_argument(
140 "input",
141 help="Target filepath to standalone line crop OR parent path containing image lists.",
142 )
143 parser.add_argument(
144 "--model",
145 default=str(HERE / "khmer_ocr_model_CRNN" / "best_model.pth"),
146 help="Checkpoint parameter file destination location path.",
147 )
148 parser.add_argument(
149 "--charset",
150 default=str(HERE / "khmer_ocr_model_CRNN" / "char.json"),
151 help="JSON configuration text format vocabulary parsing schema.",
152 )
153 parser.add_argument(
154 "--output",
155 default=str(HERE / "predictions.txt"),
156 help="Target text document destination to record string output arrays.",
157 )
158 parser.add_argument(
159 "--device",
160 default="cuda" if torch.cuda.is_available() else "cpu",
161 help="Hardware execution pipeline override runtime flag.",
162 )
163 args = parser.parse_args()
164
165 device = torch.device(args.device)
166
167 # Initialize vocabulary bounds
168 idx2char = load_charset(args.charset)
169 num_classes = max(idx2char.keys()) + 1
170
171 # Instantiate weights mapping sequence layout
172 model = ResNetBiLSTMCTC(num_classes=num_classes)
173 state_dict = torch.load(args.model, map_location="cpu")
174 model.load_state_dict(state_dict)
175 model.to(device).eval()
176
177 print(f"Model File : {args.model}")
178 print(f"Charset File : {args.charset} ({num_classes} distribution classes)")
179 print(f"Target Device: {device}\n")
180
181 input_path = Path(args.input)
182 out_path = Path(args.output)
183
184 # ── Path Evaluator Logic: Independent Image Evaluation ──────────────────
185 if input_path.is_file():
186 text = predict_one(model, input_path, idx2char, device)
187 print(f"Image File : {input_path.name}")
188 print(f"Prediction : {text}")
189 with open(out_path, "w", encoding="utf-8") as f:
190 f.write(f"{input_path.name}\t{text}\n")
191 print(f"Saved logs : {out_path}")
192 return
193
194 # ── Path Evaluator Logic: Directory Iteration Loop ───────────────────────
195 if input_path.is_dir():
196 images = sorted(
197 p for p in input_path.iterdir()
198 if p.suffix.lower() in IMAGE_EXTS
199 )
200 if not images:
201 print(f"Termination: No valid image file variations found inside directory context '{input_path}'")
202 return
203
204 print(f"Queued Processing Run: Found {len(images)} sequence elements inside directory structure.\n")
205 with open(out_path, "w", encoding="utf-8") as f:
206 for i, img_path in enumerate(images, 1):
207 try:
208 text = predict_one(model, img_path, idx2char, device)
209 except Exception as error_exception:
210 text = f"RUNTIME ERROR METRIC EXCEPTION: {error_exception}"
211 print(f"[{i}/{len(images)}] {img_path.name} -> {text}")
212 f.write(f"{img_path.name}\t{text}\n")
213
214 print(f"\nExecution Pipeline Complete: Stream saved down cleanly into '{out_path}'")
215 return
216
217 print(f"Invalid Operation Error: Resource tracking index target reference '{input_path}' does not point to structural nodes.")
218
219
220if __name__ == "__main__":
221 main()