Repository navigation
Expand file tree
/
Copy pathletterread.py
More file actions
2268 lines (1971 loc) · 115 KB
/
Copy pathletterread.py
File metadata and controls
2268 lines (1971 loc) · 115 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Learned Greek letter reader for 2D images of scroll text (ink maps, ink labels, photographs).
A small convolutional line reader, trained with CTC on synthetic lines
(scripts/letters/train_linenet.py), runs here in numpy with no torch. Each text line becomes
per-position letter probabilities, decoded into letters with boxes and confidences, with or
without a separate letter language model (kit/greeklm.py). Every string it returns is model
output: a hypothesis for a person to check, never a reading of a papyrus.
Controls travel with every result, and each can fail:
- nulls: the same reader on tile-shuffled and phase-randomized copies of the image. The
confidence threshold is set so the nulls give few letters, and the result says how many
letters the real image kept above it against each null;
- control map (optional): a second image of the same surface, such as its reverse-depth ink map,
read on the same lines at the same confidence floor;
- status: one code for the whole read and one per column (strong, weak, none, too_few or
no_control; control_status), the ratio to the largest null, and per line the count expected by
chance from the nulls, so a person or a pipeline can act on the controls;
- orientation: text the right way round should keep more letters as stored than mirrored or
turned 90, 180 or 270 degrees (weak for Greek capitals, many of which are symmetric);
- language off: the ink-only decode is always reported, and a language-model decode, when
asked for, sits beside it and never replaces it.
Photographs: polarity "auto" reads the image quickly both ways, as bright ink and as dark ink, and
keeps the polarity whose letters stand further above its own nulls (polarity_trial; gray levels
alone fail on a photograph of papyrus on a white ground); a read in a given polarity that the
controls do not pass gets the same quick check and a hint when the other polarity reads clearly
better. JPEG and PNG files are turned by their EXIF orientation as they load, read(rotate=...,
mirror=...) reads a turned image and maps every box back to input pixels, and the overlay is drawn
on the image the way round and in the polarity it was given (in color for a color photograph).
Weights are a .npz exported by the training script (format linenet-v1, BatchNorm folded into
the convolutions). They are not in git: pass a file, or the name of a released model in
kit/data/letter-weights.json, which resolve_weights() downloads once and checks by SHA-256.
Python use: read(image, "letters-v2-ccby", letter_px=86) takes a 2D array (an RGB or RGBA array is
averaged to gray), a released model name, a path or a LineNet, checks its options and returns a
JSON-able dict. Every result carries the schema number SCHEMA, the reader's version, the weights'
released name, file name and SHA-256, the seed and the options used (docs/letter-reader.md, "The
result, field by field"); no result holds a local folder.
Big images: load_reduced() block-averages an image over MAX_PIXELS by an integer factor while
loading it in row bands (memory maps, zarr, or the stored integer type; never a float copy of the
whole image), so a whole native ink map can be read; analyze() and read() take that factor as
input_scale and report every row, column, box and letter height in input pixels.
"""
import hashlib
import inspect
import io
import json
import math
import os
import sys
from pathlib import Path
import numpy as np
from . import __version__ as KIT_VERSION
NOTE = ("Model output: letter hypotheses from a learned line reader, not a reading. "
"A papyrologist decides what is legible.")
FORMAT = "linenet-v1"
# The result's schema number: raised when a field is renamed or removed or changes its meaning; new fields
# keep it. docs/letter-reader.md ("The result, field by field") lists every field of schema 1.
SCHEMA = 1
WEIGHTS_REGISTRY = Path(__file__).parent / "data" / "letter-weights.json"
# The offline and public-map demos use permissive weights. The CLI selects by task through letterweights;
# Python read() still requires an explicit model. Both released checkpoints keep their existing hashes.
DEFAULT_WEIGHTS = "letters-v2-ccby"
# Weight of the language model in its beam decode when a model is given without a weight (the command line's
# --lm-weight default).
LM_WEIGHT = 0.5
# Default line pitch in letter heights (6.6 / 3.5, about 1.89), used only when the image gives no
# line pitch of its own.
PITCH_PER_H = 6.6 / 3.5
DOWNLOAD_TIMEOUT = 120 # seconds without data before a weights download gives up
# Pixel budget. An image with more pixels than this is block-averaged while it loads (load_reduced):
# 64 Mpx is 256 MB as float32, and normalizing it takes a few such copies, which fits a laptop.
MAX_PIXELS = 64_000_000
BAND_PIXELS = 64_000_000 # input pixels per band read while block-averaging (rounded up to whole tiles)
# Smallest letter height read, in pixels of the image read: a quarter of the released networks' 16 px working
# height (LineNet.letter_px). The reader enlarges the image by 16 / letter_px each way, so a smaller letter
# costs over 16 times the pixels of the image, and its pen stroke (0.14 of the letter's height in the demo's
# hand, kit/letterdemo.py PEN) is under a pixel wide: no letter shape is left to read.
MIN_LETTER_PX = 4
# Most null copies one read makes: each is one more read of the whole image, and the status compares
# with the largest null count, so more nulls cost time and make the test stricter (docs/letter-reader.md).
MAX_NULLS = 64
OVERLAY_MAX_SIDE = 4000
NO_LETTER_SIZE = ("no letter size: the image shows no regular line spacing to estimate it from. Give the letter "
"height with --letter-px H (image pixels), or with --letter-mm MM and --um-per-px UM (for "
"example --letter-mm 3.3 --um-per-px 38.4), or read a window that holds several lines with "
"--crop Y0 Y1 X0 X1")
# Controls. A read (or a column) needs at least MIN_LETTERS confident letters before its counts can
# pass, and more than STRONG_FACTOR times every control's count (the largest null, and the control
# map when one is read) to be "strong"; docs/letter-reader.md says why these values.
MIN_LETTERS = 10
STRONG_FACTOR = 3
STATUS_TEXT = {
"strong": "Confident letters more than three times every control: worth a person's check, still model output "
"and not a reading.",
"weak": "Confident letters above every control, but not three times above all of them: weak evidence.",
"none": "Confident letters NOT above the controls: treat every letter here as noise.",
"too_few": "Too few confident letters to judge (fewer than {min_letters}).",
"no_control": "No control run (nulls=0): nothing here can be trusted.",
}
# Deskew: the largest shear angle searched, degrees either way, in steps of DESKEW_STEP.
DESKEW_DEG = 6.0
DESKEW_STEP = 0.5
# Polarity "auto" in read(), analyze() and the command line (polarity_trial): a quick read both ways.
AUTO_RULE = ("auto: the image is read quickly as bright ink and as dark ink (one step of the scale search: up to 8 "
"lines at the letter size, 2 nulls each), and the polarity whose letters kept exceed its own largest "
"null by more is used; bright on a tie")
# normalize(a, "auto") alone, without a reader (resolve_polarity): a gray-level guess, kept for callers that
# normalize without reading. It fails on a photograph of papyrus on a white ground, whose median is the papyrus.
POLARITY_RULE = ("gray levels: dark ink when the median gray lies in the upper half of the range from the 0.5th to "
"the 99.8th percentile (a light ground), else bright ink (a dark ground)")
# Orientation reads: name -> (rotate, mirror) of orient(), applied to the image as read.
ORIENTATIONS = {"as_is": (0, False), "mirrored": (0, True), "rot180": (180, False), "rot90": (90, False),
"rot270": (270, False)}
ROTATIONS = (0, 90, 180, 270)
# ------------------------------------------------------------------ released weights
class WeightsError(OSError):
"""Released weights could not be downloaded; the message names the model, the URL and the cache folder."""
def _env_weights_file():
"""The file $KIT_LETTER_WEIGHTS names (~ expanded), when it names a file rather than a folder, else None.
It names a file when it is an existing file, or a path ending in .npz."""
env = os.environ.get("KIT_LETTER_WEIGHTS")
if not env:
return None
p = Path(env).expanduser()
return p if p.is_file() or p.suffix.lower() == ".npz" else None
def weights_cache():
"""Folder for downloaded weights: $KIT_LETTER_WEIGHTS (~ expanded), else ~/.cache/scrolls-kit/letters.
When $KIT_LETTER_WEIGHTS names a weights file, the folder is the one that holds it (resolve_weights
uses that file itself when its SHA-256 is the released model's)."""
f = _env_weights_file()
if f is not None:
return f.parent
env = os.environ.get("KIT_LETTER_WEIGHTS")
return Path(env).expanduser() if env else Path.home() / ".cache" / "scrolls-kit" / "letters"
def released_weights(registry=None):
"""{name: entry} of released reader weights (file, sha256, url, licence, notes)."""
data = json.loads(Path(registry or WEIGHTS_REGISTRY).read_text(encoding="utf-8"))
return data["models"]
def released_name(sha256, registry=None):
"""The released model name whose SHA-256 is sha256, or None (also when the registry cannot be read)."""
if not sha256:
return None
try:
models = released_weights(registry)
except (OSError, ValueError, KeyError, TypeError):
return None
return next((name for name, m in sorted(models.items()) if m.get("sha256") == sha256), None)
def _sha256(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for block in iter(lambda: f.read(1 << 20), b""):
h.update(block)
return h.hexdigest()
def _by_hand(name, m, folder):
"""The sentence that says how to put a released model's file into the cache folder by hand."""
return (f"download {m['file']} by hand (from {m.get('url') or 'the address in docs/letter-reader.md'}, in a "
f"browser or with: curl -L -o {m['file']} URL) and copy it into {folder}; the reader checks its SHA-256 "
f"({m['sha256']}) there as for a download. Set KIT_LETTER_WEIGHTS to another folder, or to the file "
"itself, to keep it elsewhere")
def resolve_weights(spec, cache=None, registry=None, download=True):
"""A weights file path from a released model name or a path.
Names come first: a string that names a released model in kit/data/letter-weights.json is that
model, even when a file of that name exists. The model is the file $KIT_LETTER_WEIGHTS names when
that is a file with the model's SHA-256 (and no cache is given); otherwise the registry's file in
the cache folder (cache, else weights_cache()), downloaded from its URL once when missing, into a
file of mode 0644. Anything else is a path (~ expanded): returned as it is, except that a file
named like a released model's file must have that model's SHA-256. A file whose SHA-256 differs
is refused (ValueError), so a damaged or swapped file is never read. A failed download raises
WeightsError (an OSError) naming the model, the URL and the cache folder, and saying that a file
downloaded by hand can be copied into that folder."""
models = released_weights(registry)
if isinstance(spec, str) and spec in models:
return _released_file(spec, models[spec], cache, download)
if not isinstance(spec, (str, os.PathLike)):
raise TypeError(f"weights {spec!r}: give a released model name ({', '.join(sorted(models))}), a path to a "
f"{FORMAT} .npz file, or a LineNet")
p = Path(spec).expanduser()
if not p.is_file():
if p.exists():
raise ValueError(f"{spec}: a folder, not a {FORMAT} weights file")
raise ValueError(f"{spec}: no such file, and not a released model ({', '.join(sorted(models))})")
for name, m in sorted(models.items()):
if p.name == m["file"] and _sha256(p) != m["sha256"]:
raise ValueError(f"{spec}: has the file name of the released model {name} but not its SHA-256 "
f"{m['sha256']}, so it is damaged or another file; download it again (python -m kit "
f"letterread IMAGE --weights {name} does), or rename it if it is weights of your own")
return p
def _released_file(name, m, cache=None, download=True):
"""The checked local file of the released model `name` (registry entry m): see resolve_weights."""
env = _env_weights_file() if cache is None else None
if env is not None and env.is_file() and _sha256(env) == m["sha256"]:
return env
folder = Path(cache).expanduser() if cache else weights_cache()
dest = folder / m["file"]
if not dest.exists():
if not download:
raise ValueError(f"{name}: not in {folder}")
if not m.get("url"):
raise ValueError(f"{name}: no download address yet; {_by_hand(name, m, folder)}")
_download(name, m, dest)
if _sha256(dest) != m["sha256"]:
raise ValueError(f"{name}: {dest} does not match its SHA-256 {m['sha256']}; delete it and retry, or "
f"{_by_hand(name, m, folder)}")
return dest
def _download(name, m, dest):
"""Download a released model's file to dest (mode 0644), keeping nothing unless its SHA-256 matches."""
import http.client
import tempfile
import urllib.request
folder = dest.parent
def failed(exc):
return WeightsError(f"could not download the weights {name} from {m['url']} into {folder}: {exc}. "
f"Check the connection and retry, or {_by_hand(name, m, folder)}")
try:
folder.mkdir(parents=True, exist_ok=True)
# A file of this process's own, so two first runs never write one file, and no partial file is
# left behind on any error; the timeout turns a stalled connection into an error.
fd, tmp = tempfile.mkstemp(dir=folder, prefix=dest.name + ".", suffix=".part")
except OSError as exc:
raise failed(exc) from exc
tmp = Path(tmp)
try:
try:
with os.fdopen(fd, "wb") as f, urllib.request.urlopen(m["url"], timeout=DOWNLOAD_TIMEOUT) as r:
for block in iter(lambda: r.read(1 << 20), b""):
f.write(block)
except (OSError, http.client.HTTPException, ValueError) as exc: # URLError, HTTPError, timeouts, cut-offs
raise failed(exc) from exc
if _sha256(tmp) != m["sha256"]:
raise ValueError(f"{name}: download from {m['url']} does not match its SHA-256; nothing kept. Retry, or "
f"{_by_hand(name, m, folder)}")
os.chmod(tmp, 0o644) # mkstemp makes the file private (0600); a cache file is for every user
os.replace(tmp, dest)
finally:
tmp.unlink(missing_ok=True)
def weights_info(net):
"""Where a LineNet's weights came from, as JSON: released name (None when not a released model), file name and SHA-256.
Only the file name is kept, never its folder, so a result never holds a local path."""
return {"name": getattr(net, "name", None), "file": getattr(net, "file", None),
"sha256": getattr(net, "sha256", None), "format": FORMAT}
def load_weights(weights):
"""A LineNet from a LineNet (returned as it is), a released model name or a path (resolve_weights)."""
if isinstance(weights, LineNet):
return weights
return LineNet(resolve_weights(weights))
# ------------------------------------------------------------------ console
def _encodes(encoding, text="ΑΩϹ·"):
"""True when `encoding` can encode Greek capitals, lunate sigma and the middle dot."""
try:
text.encode(encoding or "ascii")
return True
except (LookupError, UnicodeError):
return False
def utf8_console(*streams):
"""Let text streams print Greek: a stream that is not a terminal, or cannot encode Greek, becomes UTF-8.
Each stream (default: sys.stdout and sys.stderr) that is not a terminal (a pipe or a file, where
Windows uses a code page such as cp1252) or whose encoding cannot encode Greek letters is
reconfigured to encoding="utf-8", errors="backslashreplace", so printing a read never fails on
a letter. A terminal that shows Greek is left as it is, and so is a stream without reconfigure()
(io.StringIO, as tests use). Returns the streams changed."""
changed = []
for s in streams or (sys.stdout, sys.stderr):
if s is None:
continue
try:
tty = bool(s.isatty())
except (AttributeError, ValueError, OSError):
tty = False
if tty and _encodes(getattr(s, "encoding", None)):
continue
reconf = getattr(s, "reconfigure", None)
if reconf is None:
continue
try:
reconf(encoding="utf-8", errors="backslashreplace")
except (AttributeError, ValueError, OSError, io.UnsupportedOperation):
continue
changed.append(s)
return changed
# ------------------------------------------------------------------ numpy network
class LineNet:
"""numpy inference of the exported CTC line reader."""
def __init__(self, path):
z = np.load(path, allow_pickle=False)
self.meta = json.loads(str(z["meta"]))
if self.meta.get("format") != FORMAT:
raise ValueError(f"{path}: not a {FORMAT} weight file")
self.w = {k: z[k].astype(np.float32) for k in z.files if k != "meta"}
self.alphabet = self.meta["alphabet"]
self.letter_px = int(self.meta["letter_px"])
self.band = tuple(self.meta["band"])
self.stride = int(self.meta["stride"])
# Provenance for every result (weights_info): the file's name, never its folder, its SHA-256 and,
# when that is a released model's, the model's name.
self.file = self.sha256 = self.name = None
if isinstance(path, (str, os.PathLike)):
self.file, self.sha256 = Path(path).name, _sha256(path)
self.name = released_name(self.sha256)
@property
def band_rows(self):
return int(round((self.band[0] + self.band[1]) * self.letter_px))
def logits(self, band):
"""band: (H, W) float32 in [0, 1] -> (T, K) logits with T = ceil(W / stride)."""
x = np.asarray(band, np.float32)[None]
W0 = x.shape[2]
Wp = int(math.ceil(W0 / self.stride) * self.stride)
if Wp != W0:
x = np.pad(x, ((0, 0), (0, 0), (0, Wp - W0)), constant_values=float(np.median(band)))
for layer in self.meta["layers"]:
op = layer["op"]
w = self.w.get(f"w{layer.get('id')}")
b = self.w.get(f"b{layer.get('id')}")
if op == "conv2d":
x = np.maximum(_conv2d(x, w, b, layer["pad"]), 0)
elif op == "maxpool":
x = _maxpool(x, *layer["k"])
elif op == "collapse":
x = np.maximum(_conv2d(x, w, b, 0), 0)[:, 0, :]
elif op == "conv1d_res":
x = x + np.maximum(_conv1d(x, w, b, layer["dilation"]), 0)
elif op == "head":
x = w[:, :, 0] @ x + b[:, None]
else:
raise ValueError(f"unknown layer op {op!r}")
return x.T
def log_probs(self, band):
lg = self.logits(band)
lg = lg - lg.max(axis=1, keepdims=True)
return lg - np.log(np.exp(lg).sum(axis=1, keepdims=True))
def _maxpool(x, ky, kx):
"""Max pooling of a (C, H, W) array over ky x kx blocks, dropping rows and columns that do not fill a block.
The same values, bit for bit for a C-contiguous x, as x[:, :H // ky * ky, :W // kx * kx].reshape(C, H // ky,
ky, W // kx, kx).max(axis=(2, 4)), taken as the elementwise maximum of the ky * kx strided slices: numpy
reduces over the two strided axes of the reshape about 20 times more slowly. LineNet always passes a
C-contiguous x (np.maximum output). For other layouts the old reduction's choice between -0.0 and +0.0
in a block whose maximum is a tie of the two depends on memory order, so only the sign of a zero can
differ, and the output here is always C-ordered."""
C, H, W = x.shape
hc, wc = H // ky * ky, W // kx * kx
out = x[:, 0:hc:ky, 0:wc:kx].copy()
for i in range(ky):
for j in range(kx):
if i or j:
np.maximum(out, x[:, i:hc:ky, j:wc:kx], out=out)
return out
def _conv2d(x, w, b, pad):
"""x: C x H x W, w: O x C x kh x kw -> O x H' x W' (stride 1)."""
O, C, kh, kw = w.shape
if pad:
x = np.pad(x, ((0, 0), (pad, pad), (pad, pad)))
x = np.ascontiguousarray(x)
_, H, W = x.shape
Ho, Wo = H - kh + 1, W - kw + 1
s = x.strides
cols = np.lib.stride_tricks.as_strided(x, (C, kh, kw, Ho, Wo), (s[0], s[1], s[2], s[1], s[2]))
return np.tensordot(w, cols, axes=([1, 2, 3], [0, 1, 2])) + b[:, None, None]
def _conv1d(x, w, b, dilation):
"""x: C x T, w: O x C x k, same padding, dilated."""
O, C, k = w.shape
p = dilation * (k - 1) // 2
xp = np.pad(x, ((0, 0), (p, p)))
T = x.shape[1]
out = np.zeros((O, T), np.float32)
for j in range(k):
out += w[:, :, j] @ xp[:, j * dilation:j * dilation + T]
return out + b[:, None]
# ------------------------------------------------------------------ CTC decoding
def ctc_greedy(lp, alphabet, blank=0):
"""Best-path decode with per-letter confidence and frame span: [(letter, p, t0, t1)]."""
best = lp.argmax(axis=1)
out = []
prev = blank
for t, k in enumerate(best):
if k != blank and k != prev:
out.append([alphabet[k - 1], float(np.exp(lp[t, k])), t, t])
elif k != blank and k == prev and out:
out[-1][1] = max(out[-1][1], float(np.exp(lp[t, k])))
out[-1][3] = t
prev = k
return [tuple(o) for o in out]
def ctc_beam(lp, alphabet, lm=None, lm_weight=0.0, beam=32, blank=0, insert_bonus=0.0, top_k=8):
"""CTC prefix beam search, optionally with shallow fusion of a letter language model.
lm: any object with logp_next(prefix) -> log-probabilities over the alphabet, in order.
Returns (best string, its log score)."""
T, K = lp.shape
beams = {"": (0.0, -np.inf)} # prefix -> (log p ending in blank, ending in a letter)
for t in range(T):
nxt = {}
top = np.argsort(-lp[t])[:top_k]
for pre, (pb, pnb) in beams.items():
tot = np.logaddexp(pb, pnb)
b_, nb_ = nxt.get(pre, (-np.inf, -np.inf))
nxt[pre] = (np.logaddexp(b_, tot + lp[t, blank]), nb_)
lmlp = lm.logp_next(pre) if (lm is not None and lm_weight) else None
for k in top:
if k == blank:
continue
c = alphabet[k - 1]
if pre and pre[-1] == c:
b_, nb_ = nxt.get(pre, (-np.inf, -np.inf))
nxt[pre] = (b_, np.logaddexp(nb_, pnb + lp[t, k]))
src = pb
else:
src = tot
new = pre + c
bonus = insert_bonus + (lm_weight * lmlp[k - 1] if lmlp is not None else 0.0)
b_, nb_ = nxt.get(new, (-np.inf, -np.inf))
nxt[new] = (b_, np.logaddexp(nb_, src + lp[t, k] + bonus))
beams = dict(sorted(nxt.items(), key=lambda kv: -np.logaddexp(*kv[1]))[:beam])
best = max(beams.items(), key=lambda kv: np.logaddexp(*kv[1]))
return best[0], float(np.logaddexp(*best[1]))
def force_align(lp, text, alphabet, blank=0):
"""CTC forced alignment (Viterbi): the frames of each letter of a known text in one line's log-probs.
Returns [(letter, t0, t1, mean p)] in text order, or None when the line has too few frames for
the text. This turns a line-level transcription into letter positions without drawing boxes."""
idx = [alphabet.index(c) + 1 for c in text]
T = lp.shape[0]
if not idx:
return []
ext = [blank]
for k in idx:
ext += [k, blank]
S = len(ext)
need = len(idx) + sum(1 for a, b in zip(idx, idx[1:]) if a == b)
if T < need:
return None
ext = np.array(ext)
em = lp[:, ext] # T x S
alpha = np.full((T, S), -np.inf)
back = np.zeros((T, S), np.int8) # 0 stay, 1 from s-1, 2 from s-2
alpha[0, 0] = em[0, 0]
alpha[0, 1] = em[0, 1]
skip = np.zeros(S, bool)
skip[2:] = (ext[2:] != blank) & (ext[2:] != ext[:-2])
for t in range(1, T):
a0 = alpha[t - 1]
a1 = np.concatenate([[-np.inf], a0[:-1]])
a2 = np.where(skip, np.concatenate([[-np.inf, -np.inf], a0[:-2]]), -np.inf)
stack = np.stack([a0, a1, a2])
back[t] = stack.argmax(axis=0)
alpha[t] = stack.max(axis=0) + em[t]
s = S - 1 if alpha[T - 1, S - 1] >= alpha[T - 1, S - 2] else S - 2
if not np.isfinite(alpha[T - 1, s]):
return None
path = np.empty(T, int)
for t in range(T - 1, -1, -1):
path[t] = s
s -= int(back[t, s])
out = []
for i, c in enumerate(text):
frames = np.nonzero(path == 2 * i + 1)[0]
out.append((c, int(frames[0]), int(frames[-1]), float(np.exp(lp[frames, idx[i]]).mean())))
return out
def locate(query, text, top=3):
"""Where a short letter string best matches inside a long text, allowing edits (semi-global alignment).
Returns up to top (start, end, edit distance) with end exclusive, best first, non-overlapping.
Used to find which published line a read line belongs to before forced alignment."""
if not query:
return []
t = np.frombuffer(text.encode("utf-32-le"), dtype=np.uint32)
q = np.frombuffer(query.encode("utf-32-le"), dtype=np.uint32)
n = len(t)
ar = np.arange(n + 1)
D = np.zeros(n + 1) # free start anywhere in the text
for ch in q:
sub = D[:-1] + (t != ch)
E = np.concatenate([[D[0] + 1], np.minimum(D[1:] + 1, sub)])
D = np.minimum.accumulate(E - ar) + ar # horizontal moves cost 1 each
found = []
order = np.argsort(D[1:], kind="stable") + 1
for end in order:
if len(found) >= top:
break
if any(abs(int(end) - e) < len(query) for _, e, _ in found):
continue
d = int(D[end])
start = _locate_start(q, t, int(end), d)
found.append((start, int(end), d))
return found
def _locate_start(q, t, end, d):
"""Start of the best match ending at end: align the reversed query to the reversed text before end."""
lo = max(0, end - len(q) - d)
seg = t[lo:end][::-1]
rq = q[::-1]
n = len(seg)
ar = np.arange(n + 1)
D = ar.astype(float) # anchored at end
for ch in rq:
sub = D[:-1] + (seg != ch)
E = np.concatenate([[D[0] + 1], np.minimum(D[1:] + 1, sub)])
D = np.minimum.accumulate(E - ar) + ar
k = int(np.argmin(D))
return end - k
def edit_distance(a, b):
prev = list(range(len(b) + 1))
for i, ca in enumerate(a, 1):
cur = [i]
for j, cb in enumerate(b, 1):
cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
prev = cur
return prev[-1]
# ------------------------------------------------------------------ image and page geometry
RGB_MEAN = (1 / 3, 1 / 3, 1 / 3) # color TIFF and .npy: the plain mean of the first three channels
RGB_LUMA = (0.299, 0.587, 0.114) # color PNG and JPEG: Pillow's gray ("F") conversion
class _Source:
"""A 2D image (or a 2D color image) read window by window in its stored type.
arr is anything with .shape and numpy slicing: an array, a memory map or a zarr array. Size-1
axes are dropped as np.squeeze would, without reading the data; chunk_rows is the height of the
source's tiles or chunks, when it has them."""
def __init__(self, arr, name, weights=RGB_MEAN, chunk_rows=None, store=None):
self.arr, self.name, self.weights, self.chunk_rows, self.store = arr, name, weights, chunk_rows, store
self.full = tuple(int(n) for n in arr.shape)
kept = tuple(n for n in self.full if n != 1)
self.rgb = len(kept) == 3 and kept[-1] in (3, 4)
if len(kept) != 2 and not self.rgb:
raise ValueError(f"{name}: expected a 2D image, got shape {kept}")
self.shape = kept[:2]
def read(self, y0, y1, x0, x1):
"""Rows y0:y1 and columns x0:x1 as a numpy array in the stored type: (h, w), or (h, w, c) in color."""
sl = [slice(y0, y1), slice(x0, x1), slice(None)]
return np.asarray(self.arr[tuple(0 if n == 1 else sl.pop(0) for n in self.full)])
def as_float(self, y0, y1, x0, x1):
"""The window as float32 gray, exactly as load_image has always converted it."""
a = self.read(y0, y1, x0, x1)
if a.ndim == 3:
a = a[..., :3].mean(axis=-1)
return a.astype(np.float32)
def close(self):
"""Close the file a TIFF's zarr view keeps open."""
if self.store is not None and hasattr(self.store, "close"):
self.store.close()
def _is_zarr(p):
p = Path(p)
return p.suffix.lower() == ".zarr" or (p.is_dir() and any((p / n).exists() for n in (".zarray", ".zgroup", "zarr.json")))
def _ome_level0(attrs):
"""Path of an OME-Zarr group's full-resolution level (its first dataset), or None."""
try:
ms = attrs.get("multiscales") or (attrs.get("ome") or {}).get("multiscales")
return str(ms[0]["datasets"][0]["path"]) if ms else None
except (AttributeError, IndexError, KeyError, TypeError):
return None
def _zarr_array(store, key, name, strict_key=True):
"""The zarr array at `key` (an array path, or an OME-Zarr level such as "0") of a store or path.
Without a key: the array itself, an OME-Zarr group's full-resolution level, or a group's only array."""
try:
import zarr
except ImportError as exc:
raise ImportError(f"{name}: reading zarr needs the zarr package (python -m pip install zarr)") from exc
z = zarr.open(store, mode="r")
if not isinstance(z, zarr.Group):
if key and strict_key:
raise ValueError(f"{name}: a single zarr array, so --key {key} does not apply")
return z
names = sorted(z.array_keys())
if not key:
key = _ome_level0(z.attrs) or (names[0] if len(names) == 1 else None)
if key is None:
raise ValueError(f"{name}: a zarr group with arrays {names or 'none'}; pick one with --key "
"(an array path, or an OME-Zarr level such as 0)")
try:
arr = z[key]
except KeyError:
raise ValueError(f"{name}: no array {key!r} in this zarr (arrays: {', '.join(names) or 'none'})") from None
if isinstance(arr, zarr.Group):
raise ValueError(f"{name}: {key!r} is a zarr group, not an array; give its level, such as {key}/0")
return arr
def _chunk_rows(arr):
chunks = getattr(arr, "chunks", None)
try:
return int(chunks[-2]) if chunks and len(chunks) >= 2 else None
except (TypeError, ValueError):
return None
def _open(path, key=None, whole=False):
"""A _Source for an image file or zarr; whole=True when every pixel will be read at full resolution.
Windows and bands are read without decoding the rest where the format allows: .npy and
uncompressed TIFF by memory map, zarr (and OME-Zarr levels) by chunk, compressed tiled TIFF
through tifffile's zarr view when zarr is installed. A compressed TIFF without zarr, a .npz and
a PNG or JPEG are decoded whole, in their stored type."""
p = Path(path)
if _is_zarr(p):
if not p.exists():
raise FileNotFoundError(f"{p}: no such file or folder")
arr = _zarr_array(str(p), key, p)
return _Source(arr, p, chunk_rows=_chunk_rows(arr))
suf = p.suffix.lower()
if suf == ".npy":
return _Source(np.load(p, allow_pickle=False, mmap_mode="r"), p)
if suf == ".npz":
z = np.load(p, allow_pickle=False)
return _Source(z[key or z.files[0]], p)
if suf in (".tif", ".tiff"):
import tifffile
if whole:
return _Source(tifffile.imread(str(p)), p)
try:
return _Source(tifffile.memmap(str(p), mode="r"), p)
except ValueError: # a compressed TIFF cannot be memory-mapped
pass
try:
import zarr # noqa: F401
except ImportError:
return _Source(tifffile.imread(str(p)), p)
store = tifffile.imread(str(p), aszarr=True)
try:
arr = _zarr_array(store, key, p, strict_key=False)
except BaseException:
store.close()
raise
return _Source(arr, p, chunk_rows=_chunk_rows(arr), store=store)
im = _pillow_image(p)
if whole:
return _Source(np.asarray(im.convert("F")), p)
if im.mode in ("1", "LA"):
im = im.convert("L")
elif im.mode not in ("L", "I", "F") and not im.mode.startswith("I;16"):
im = im.convert("RGB")
return _Source(np.asarray(im), p, weights=RGB_LUMA)
def _is_pillow(p):
return not _is_zarr(p) and Path(p).suffix.lower() not in (".npy", ".npz", ".tif", ".tiff")
EXIF_ORIENTATION_TAG = 0x0112
def _exif_tag(im):
"""The EXIF orientation of an open Pillow image: 1 (as stored) to 8; 1 when it has none or it is unreadable."""
try:
n = int(im.getexif().get(EXIF_ORIENTATION_TAG, 1))
except Exception: # a damaged EXIF block must not stop the read; the pixels are then read as stored
return 1
return n if 1 <= n <= 8 else 1
def exif_orientation(path):
"""EXIF orientation tag of a PNG or JPEG file (1 to 8; 5 to 8 swap height and width), 1 for other formats.
The reader applies it as it loads such a file, so input pixels are the image as a viewer shows it."""
if not _is_pillow(path):
return 1
from PIL import Image
Image.MAX_IMAGE_PIXELS = None
with Image.open(path) as im:
return _exif_tag(im)
def _pillow_image(p):
"""A PNG, JPEG or other Pillow image, turned by its EXIF orientation (ImageOps.exif_transpose) when it has one."""
from PIL import Image, ImageOps
Image.MAX_IMAGE_PIXELS = None
im = Image.open(p)
if _exif_tag(im) != 1:
im = ImageOps.exif_transpose(im)
return im
def _pillow_shape(p):
"""(height, width) of a Pillow image as a viewer shows it (EXIF orientation applied), without decoding its pixels."""
from PIL import Image
Image.MAX_IMAGE_PIXELS = None
with Image.open(p) as im:
return (im.width, im.height) if _exif_tag(im) in (5, 6, 7, 8) else (im.height, im.width)
def image_shape(path, key=None):
"""(height, width) of a whole image as the reader loads it, without decoding its pixels where the format allows.
A PNG or JPEG is measured the way round a viewer shows it (EXIF orientation applied); zarr and
.npz: the array at key (load_image's default without one)."""
if _is_pillow(path):
return tuple(_pillow_shape(path))
src = _open(path, key)
try:
return tuple(int(n) for n in src.shape)
finally:
src.close()
def display_image(path, crop=None, k=1):
"""A color PNG or JPEG as an RGB uint8 array for the overlay: EXIF orientation applied, cropped, block-averaged by k.
The shape is that of load_reduced's image for the same crop and k. None for a gray image and for
other formats, whose overlay is drawn from the read image in its own polarity."""
if not _is_pillow(path):
return None
im = _pillow_image(path)
if im.mode in ("1", "L", "LA", "I", "F") or im.mode.startswith("I;"):
return None
y0, y1, x0, x1 = _window(crop, (im.height, im.width), path)
ny, nx = (y1 - y0) // k, (x1 - x0) // k
im = im.convert("RGB").crop((x0, y0, x0 + nx * k, y0 + ny * k))
if k > 1:
im = im.reduce(k)
return np.asarray(im)
def _window(crop, shape, name):
"""(y0, y1, x0, x1) of a crop inside an image of this shape, with numpy's slicing rules."""
H, W = shape
if not H or not W:
raise ValueError(f"{name}: the image is empty ({H} x {W} px)")
if not crop:
return 0, H, 0, W
y0, y1, x0, x1 = crop
ys, xs = range(H)[slice(y0, y1)], range(W)[slice(x0, x1)]
if not len(ys) or not len(xs):
raise ValueError(f"{name}: crop {list(crop)} is empty for an image of shape {tuple(shape)}")
return ys.start, ys.stop, xs.start, xs.stop
def load_image(path, key=None, crop=None):
"""2D float array from .npy, .npz (first or named array), .tif/.tiff (tifffile), .png/.jpg (Pillow) or zarr.
A PNG or JPEG with an EXIF orientation tag is turned by it first, so its pixels are the image as
a viewer shows it (exif_orientation() gives the tag).
crop: (Y0, Y1, X0, X1), applied before the conversion to float; a .npy file and an uncompressed
TIFF are memory-mapped, and a zarr or (with zarr installed) a compressed tiled TIFF is read by
chunk, so only the window is read. zarr: key is the array path, or the OME-Zarr level ("0" is
full resolution, the default)."""
src = _open(path, key, whole=not crop or _is_pillow(path)) # Pillow images: always its gray ("F") conversion
try:
return src.as_float(*_window(crop, src.shape, path))
finally:
src.close()
def reduce_factor(shape, letter_px=None, net_letter_px=16, max_pixels=MAX_PIXELS):
"""Integer block size for reading an image of this shape: 1 within the pixel budget.
Over the budget with a known letter height: the largest factor that keeps letters at least
twice the network's working letter height (net_letter_px), so the reader's own resize still
averages several pixels into each of its pixels; with small letters that can leave the image
over the budget. Over the budget with no letter height: the smallest factor that fits it."""
n = int(shape[0]) * int(shape[1])
if n <= max_pixels:
return 1
if letter_px:
return max(1, int(float(letter_px) // (2 * net_letter_px)))
return max(1, math.ceil(math.sqrt(n / max_pixels)))
def _block_mean(src, y0, y1, x0, x1, k, progress=None, band_pixels=BAND_PIXELS):
"""Block mean by k of src's window, float32, read in row bands (rows left over from the last block are dropped).
Bands follow the source's tiles, so each tile is decoded once; the rows of a band that do not
fill a block are carried into the next band."""
ny, nx = (y1 - y0) // k, (x1 - x0) // k
out = np.empty((ny, nx), np.float32)
yend, xend = y0 + ny * k, x0 + nx * k
rows = max(k, band_pixels // (8 if src.rgb else 1) // max(1, xend - x0)) # color is summed in float64
if src.chunk_rows:
rows = max(1, -(-rows // src.chunk_rows)) * src.chunk_rows
done, carry, r, tick = 0, None, y0, y0
while r < yend:
r1 = min(yend, (r // rows + 1) * rows)
part = src.read(r, r1, x0, xend)
if src.rgb:
part = sum(part[..., c].astype(np.float64) * w for c, w in enumerate(src.weights))
if carry is not None:
part = np.concatenate([carry, part])
n = part.shape[0] // k
if n:
out[done:done + n] = _blocks(part[:n * k], k, nx)
done += n
carry = part[n * k:].copy()
r = r1
if progress and (r - tick >= 0.1 * (yend - y0) or r >= yend):
tick = r
progress(f"block-averaging: rows {r - y0} of {y1 - y0}")
return out
def _blocks(part, k, nx, piece_bytes=64 << 20):
"""Block means by k of a band k * n rows tall and k * nx columns wide, summed in float64 a few rows at a time."""
n = part.shape[0] // k
out = np.empty((n, nx), np.float32)
step = max(1, piece_bytes // (8 * k * nx * k))
for i in range(0, n, step):
piece = part[i * k:(i + step) * k]
s = piece.reshape(-1, k, nx, k).sum(axis=3, dtype=np.float64).sum(axis=1)
out[i:i + step] = s / (k * k)
return out
def load_reduced(path, key=None, crop=None, letter_px=None, net_letter_px=16, max_pixels=MAX_PIXELS, progress=None):
"""An image for reading, block-averaged while it loads when it is over the pixel budget.
Returns (image, k, shape): the float32 image, the integer block size k (reduce_factor; 1 when
the image fits the budget, and then the image is exactly load_image's) and the (H, W) of the
input window in input pixels. Pass k to analyze() or read() as input_scale, with letter_px and
lines in input pixels; they report rows, columns, boxes and letter heights in input pixels.
The crop is applied in input pixels, before the reduction, so boxes stay relative to Y0, X0."""
p = Path(path)
if _is_pillow(p): # Pillow reads the size without decoding; the pixels are decoded once, below
shape, src = _pillow_shape(p), None
else:
src = _open(p, key)
shape = src.shape
try:
y0, y1, x0, x1 = _window(crop, shape, path)
h, w = y1 - y0, x1 - x0
k = min(reduce_factor((h, w), letter_px, net_letter_px, max_pixels), h, w)
if k == 1:
if progress and h * w > max_pixels:
progress(f"{p.name}: {h} x {w} px is over the {max_pixels / 1e6:g} Mpx budget, but its letters are "
"too small to block-average; loading it whole")
img = load_image(p, key, crop) if src is None else src.as_float(y0, y1, x0, x1)
return img, 1, (h, w)
if progress:
why = (f"letters {float(letter_px) / k:.1f} px tall after it" if letter_px else
f"letter size unknown, so only enough to fit the budget; if letters are under "
f"{2 * net_letter_px * k} px tall in the input, give their size")
progress(f"{p.name}: {h} x {w} px is over the {max_pixels / 1e6:g} Mpx budget: block-averaging by {k} "
f"while loading ({why})")
if src is None:
src = _open(p, key)
return _block_mean(src, y0, y1, x0, x1, k, progress), k, (h, w)
finally:
if src is not None:
src.close()
def normalize(a, polarity="bright", dtype=np.float64):
"""Robust 0..1 scaling with ink bright. polarity: bright (ink maps), dark (photographs) or auto.
dtype is the arithmetic precision; the result is always float32. normalize_for_reading uses
float32 arithmetic for images over the pixel budget, to halve the memory. auto here is the
gray-level guess POLARITY_RULE (resolve_polarity), for callers without a reader; read(),
analyze() and the command line decide auto with quick reads instead (polarity_trial).
normalize_info() also says which polarity was used."""
return normalize_info(a, polarity, dtype)[0]
def resolve_polarity(lo, med, hi, polarity="auto"):
"""The polarity ("bright" or "dark") from the image's 0.5th, 50th and 99.8th percentiles: POLARITY_RULE.
bright and dark are kept as asked. auto is dark when the median lies in the upper half of the
range (more than halfway from lo to hi), as when a light ground fills most of the image, else
bright. This is normalize()'s guess without a reader; it fails on photographs of papyrus on a
white ground (the white fills the top of the range and the brown papyrus sits in its lower half),
which is why reads decide auto with polarity_trial."""
if polarity not in ("bright", "dark", "auto"):
raise ValueError(f"polarity {polarity!r}: give bright, dark or auto")
if polarity != "auto":
return polarity
return "dark" if (med - lo) > (hi - med) else "bright"
def normalize_info(a, polarity="bright", dtype=np.float64):
"""normalize() and the polarity it used: (image, info).
info: asked (the polarity given), used (bright or dark), background (where the median gray lies
between the 0.5th and 99.8th percentiles, 0 to 1, None for a constant image; a description of
the image, not a rule) and, for auto, rule (POLARITY_RULE)."""
a = np.asarray(a, dtype=dtype)
finite = np.isfinite(a)
# Only then is the fill needed: the median of a full image costs a copy and a partition. An empty image
# still raises the ValueError below, and a bool image still becomes float through np.where.
if a.size == 0 or a.dtype == bool or not finite.all():
if not finite.any():
raise ValueError("image has no finite values")
a = np.where(finite, a, np.median(a[finite]))
lo, med, hi = np.percentile(a, [0.5, 50, 99.8])
if hi <= lo: # sparse binary or quantized image, such as an ink label with under 0.2% ink
lo, hi = float(a.min()), float(a.max())
info = {"asked": polarity}
if polarity == "auto":
info["rule"] = POLARITY_RULE
if hi <= lo:
info.update(used=resolve_polarity(lo, med, hi, polarity) if polarity != "auto" else "bright", background=None)
return np.zeros_like(a, dtype=np.float32), info
used = resolve_polarity(lo, med, hi, polarity)
info.update(used=used, background=round(float((med - lo) / (hi - lo)), 3))
a = np.clip((a - lo) / (hi - lo), 0, 1)
return (1.0 - a if used == "dark" else a).astype(np.float32), info
def normalize_for_reading(a, polarity="bright", max_pixels=MAX_PIXELS, with_info=False):
"""normalize() once for a whole read: float64 arithmetic within the pixel budget (as analyze() does), float32 above it.
read() and the command line normalize once and hand the result to the main read, its nulls, the
orientation reads and the overlay (each with normalized=True). with_info=True returns
(image, info) as normalize_info() does."""
out = normalize_info(a, polarity, dtype=np.float64 if np.size(a) <= max_pixels else np.float32)
return out if with_info else out[0]
def polarity_trial(a, weights, polarity="bright", *, letter_px=None, input_scale=1, seed=0, columns="auto",
deskew_deg=DESKEW_DEG, lines=None, rotate=0, mirror=False, letter_px_fallback=True,
min_letters=MIN_LETTERS):
"""Quick reads of a normalized image both ways, as it is and inverted: which polarity shows letters above its nulls.
a is normalized in `polarity` (ink bright in the array either way); it is read as given and as
1 - a, the other polarity, each once at the working scale like one step of the scale search (up
to 8 lines, 2 nulls with seed + 1, the confidence floor from those nulls), after rotate and
mirror and on the given lines (input pixels, in the turned image) when there are any. letter_px
is in input pixels; None takes the line-pitch estimate of the turned image (estimate_letter_px;
without a line pitch that raises ValueError(NO_LETTER_SIZE) when letter_px_fallback is False).
A dark-ink read fills rows the deskew shears in with its median gray, as read() does.
Returns {"rule", "letter_px", "bright": {...}, "dark": {...}, "choice"}: per polarity the letters
kept, the nulls' counts, the margin (kept minus the largest null, as the scale search ranks)
and the status those counts give (control_status, min_letters); choice is the polarity with the