""" Synthetic Handwritten Paragraph Generator Single-Writer Consistency | Cross-Source Mixing | Zero Duplicate Orders Generates synthetic paragraph images from handwritten line sources for pre-training paragraph recognition models. Supports RTL scripts. Guarantees: 1. Single-writer consistency: all lines in each paragraph from one writer 2. Cross-source mixing: multi-line paragraphs use lines from 2+ sources 3. Zero duplicate text orderings across the entire dataset 4. Source-level isolation: training and validation use separate line pools 5. Configurable reuse caps per source to control line repetition Usage: python generate_paragraphs.py \ --unique_train_dir ./data/UniqueLines/Training \ --fixed_train_dir ./data/FixedLines/Training \ --synthetic_train_dir ./data/SyntheticLines/Training \ --unique_val_dir ./data/UniqueLines/Validation \ --fixed_val_dir ./data/FixedLines/Validation \ --synthetic_val_dir ./data/SyntheticLines/Validation \ --output_dir ./SyntheticParagraphs_12000 \ --dataset_size 12000 """ import os, glob, random, argparse, gc import numpy as np from PIL import Image from tqdm import tqdm from datetime import datetime from collections import defaultdict def parse_args(): p = argparse.ArgumentParser(description="Synthetic Paragraph Generator") p.add_argument("--unique_train_dir", type=str, required=True) p.add_argument("--fixed_train_dir", type=str, default=None) p.add_argument("--synthetic_train_dir", type=str, default=None) p.add_argument("--unique_val_dir", type=str, required=True) p.add_argument("--fixed_val_dir", type=str, default=None) p.add_argument("--synthetic_val_dir", type=str, default=None) p.add_argument("--output_dir", type=str, required=True) p.add_argument("--output_format", type=str, default="TIFF", choices=["TIFF","PNG","JPEG"]) p.add_argument("--dataset_size", type=int, default=12000) p.add_argument("--train_ratio", type=float, default=0.85) p.add_argument("--min_lines", type=int, default=1) p.add_argument("--max_lines", type=int, default=7) p.add_argument("--spacing_min", type=int, default=15) p.add_argument("--spacing_max", type=int, default=35) p.add_argument("--canvas_width", type=int, default=2470) p.add_argument("--canvas_height", type=int, default=1200) p.add_argument("--padding", type=int, default=40) p.add_argument("--train_fixed_cap", type=float, default=1.5) p.add_argument("--train_synthetic_cap", type=float, default=2.5) p.add_argument("--val_fixed_cap", type=float, default=1.0) p.add_argument("--val_synthetic_cap", type=float, default=2.0) p.add_argument("--crop_whitespace", action="store_true", default=True) p.add_argument("--no_crop_whitespace", action="store_true") p.add_argument("--clean_left_edge", action="store_true", default=True) p.add_argument("--no_clean_left_edge", action="store_true") p.add_argument("--whitespace_threshold", type=int, default=250) p.add_argument("--edge_pixels", type=int, default=8) p.add_argument("--max_attempts", type=int, default=500) p.add_argument("--seed", type=int, default=42) p.add_argument("--gc_interval", type=int, default=100) return p.parse_args() def extract_writer_id(filename): basename = os.path.splitext(os.path.basename(filename))[0] parts = basename.split('_') return parts[0] if parts else basename def load_line_dataset(directory, name="Dataset"): if not directory or not os.path.exists(directory): return [] files = [] for ext in ["*.tif","*.tiff","*.png","*.jpg","*.jpeg","*.bmp"]: files.extend(glob.glob(os.path.join(directory, ext))) files.extend(glob.glob(os.path.join(directory, ext.upper()))) files = sorted(list(set(files))) data, skipped = [], 0 for p in files: lp = os.path.splitext(p)[0] + ".txt" if not os.path.exists(lp): skipped += 1; continue try: with open(lp, "r", encoding="utf-8") as f: label = f.readline().strip() except: try: with open(lp, "r", encoding="utf-8-sig") as f: label = f.readline().strip() except: skipped += 1; continue if label: data.append((p, label)) print(f" {name}: {len(data)} lines, {skipped} skipped") return data def merge_lines_by_writer(source_pairs): merged = defaultdict(list) counts = defaultdict(lambda: defaultdict(int)) for src, lines in source_pairs: for path, label in lines: wid = extract_writer_id(path) merged[wid].append((path, label, src)) counts[wid][src] += 1 return dict(merged), {k: dict(v) for k, v in counts.items()} class SingleWriterParagraphGenerator: def __init__(self, merged, totals, fixed_cap, synth_cap, min_l, max_l, max_att): self.merged = merged self.used = set() self.totals = totals self.usage = defaultdict(int) self.fcap, self.scap = fixed_cap, synth_cap self.min_l, self.max_l, self.max_att = min_l, max_l, max_att self.line_usage = defaultdict(int) self.line_src = {} for lines in merged.values(): for p, _, s in lines: self.line_src[p] = s self.valid = {w: l for w, l in merged.items() if len(l) >= min_l} self.wlist = list(self.valid.keys()) self.writers_used = set() self.src_para = defaultdict(int) self.n_paras = 0 self.n_lines = 0 self.n_dups = 0 def _capped(self, s): if s == "unique": return False c = self.fcap if s == "fixed" else self.scap return self.usage[s] >= int(c * self.totals.get(s, 0)) def _both_capped(self): return self._capped("fixed") and self._capped("synthetic") def _avail(self, wlines): return [x for x in wlines if x[2] == "unique" or not self._capped(x[2])] def get_paragraph_lines(self): if not self.wlist: return None bc = self._both_capped() for _ in range(self.max_att): wid = random.choice(self.wlist) av = self._avail(self.valid[wid]) if len(av) < self.min_l: continue nl = random.randint(self.min_l, min(self.max_l, len(av))) sel = None if nl >= 2 and not bc: srcs = set(s for _, _, s in av) if len(srcs) < 2: if self.min_l <= 1: nl = 1; sel = random.sample(av, 1) else: continue else: for _ in range(30): c = random.sample(av, nl) if len(set(s for _, _, s in c)) >= 2: sel = c; break if sel is None: continue else: sel = random.sample(av, nl) if sel is None: continue key = tuple(l for _, l, _ in sel) if key in self.used: self.n_dups += 1; continue tmp = defaultdict(int) for _, _, s in sel: if s in ("fixed", "synthetic"): tmp[s] += 1 ok = True for s in ("fixed", "synthetic"): if tmp[s] > 0: c = self.fcap if s == "fixed" else self.scap if self.usage[s] + tmp[s] > int(c * self.totals.get(s, 0)): ok = False; break if not ok: continue self.used.add(key); self.n_paras += 1; self.n_lines += nl self.writers_used.add(wid) si = set() for p, _, s in sel: self.usage[s] += 1; self.line_usage[p] += 1; si.add(s) for s in si: self.src_para[s] += 1 return sel, wid return None def get_stats(self): st = {} for s in ["unique", "fixed", "synthetic"]: t = self.totals.get(s, 0); u = self.usage.get(s, 0) uu = sum(1 for p, c in self.line_usage.items() if c > 0 and self.line_src.get(p) == s) st[s] = {'available': t, 'used': u, 'unique_used': uu, 'ratio': u / max(t, 1), 'utilisation': uu / max(t, 1) * 100} return st def load_image(path): try: return Image.open(path).convert("RGB") except: return None def crop_whitespace(image, threshold=250, margin=5): g = np.array(image.convert('L')) m = g < threshold r, c = np.any(m, axis=1), np.any(m, axis=0) if not np.any(r) or not np.any(c): return image ri, ci = np.where(r)[0], np.where(c)[0] return image.crop((max(0, ci[0]-margin), max(0, ri[0]-margin), min(image.width, ci[-1]+margin+1), min(image.height, ri[-1]+margin+1))) def clean_left_edge(image, edge_px=8, wt=240, rs=10, vt=500): a = np.array(image, dtype=np.float32) _, w = a.shape[:2] if w <= edge_px: return image for col in range(min(edge_px, w)): cd = a[:, col, :] rm, gm, bm = np.mean(cd[:,0]), np.mean(cd[:,1]), np.mean(cd[:,2]) ov = (rm+gm+bm)/3; v = np.var(cd) if (ov > wt or (rm > gm+rs and rm > bm+rs) or (v < vt and ov > 180) or (rm > 200 and rm > gm and rm > bm and ov > 180)): a[:, col, :] = 255.0 return Image.fromarray(a.astype(np.uint8)) def process_line(img, cw, do_crop, do_clean, wst, epx): if do_crop: img = crop_whitespace(img, threshold=wst, margin=3) if do_clean: img = clean_left_edge(img, edge_px=epx) if img.width > cw: s = cw / img.width img = img.resize((cw, max(int(img.height * s), 20)), Image.Resampling.LANCZOS) return img def create_paragraph(imgs, sp, cw, ch, pad, content_w, do_crop, do_clean, wst, epx): proc = [process_line(i, content_w, do_crop, do_clean, wst, epx) for i in imgs if i.width > 0 and i.height > 0] if not proc: return None, 0 th = pad*2 + sum(p.height for p in proc) + sp*(len(proc)-1) ah = min(th, ch) canvas = Image.new('RGB', (cw + pad*2, ah), (255, 255, 255)) y, used = pad, 0 for p in proc: if y + p.height > ah - pad: break x = max(cw + pad - p.width, pad) canvas.paste(p, (x, y)); y += p.height + sp; used += 1 return canvas, used def generate_split(gen, n, out_dir, name, cw, ch, pad, smin, smax, do_crop, do_clean, wst, epx, fmt, gc_int): os.makedirs(out_dir, exist_ok=True) content_w = cw - pad*2 wc = defaultdict(int); ld = defaultdict(int) cnt, err = 0, 0 pbar = tqdm(range(n), desc=f"Generating {name}") for i in pbar: try: r = gen.get_paragraph_lines() if r is None: err += 1; continue sel, wid = r imgs = [(load_image(p), l) for p, l, _ in sel] imgs = [(im, l) for im, l in imgs if im is not None] if not imgs: err += 1; continue sp = random.randint(smin, smax) pi, lu = create_paragraph([im for im, _ in imgs], sp, cw, ch, pad, content_w, do_crop, do_clean, wst, epx) if pi is None or lu == 0: err += 1; continue cnt += 1; wc[wid] += 1; ld[lu] += 1 ext = {"TIFF": "tif", "PNG": "png", "JPEG": "jpg"}[fmt] pi.save(os.path.join(out_dir, f"{wid}_para_{wc[wid]:04d}.{ext}"), fmt) with open(os.path.join(out_dir, f"{wid}_para_{wc[wid]:04d}.txt"), "w", encoding="utf-8") as f: f.write("\n".join(l for _, l in imgs[:lu])) del pi if (i+1) % gc_int == 0: gc.collect() pbar.set_postfix({"saved": cnt, "err": err}) except Exception as e: err += 1 if err < 10: print(f"\nError: {e}") gc.collect() gc.collect() print(f" {name}: {cnt:,} saved, {err:,} errors") return cnt, err, dict(ld) def print_stats(name, gen, count, ld): st = gen.get_stats() tl = sum(k*v for k, v in ld.items()) tp = sum(ld.values()) print(f"\n {name}:") print(f" Paragraphs: {count:,}, Writers: {len(gen.writers_used)}") if tp > 0: print(f" Avg lines/para: {tl/tp:.2f}") for s in ["unique", "fixed", "synthetic"]: d = st[s] if d['available'] > 0: print(f" {s.capitalize():12s}: {d['ratio']:.2f}x reuse, " f"{d['unique_used']:,}/{d['available']:,} ({d['utilisation']:.1f}%)") print(f" Duplicates rejected: {gen.n_dups:,}") def main(): args = parse_args() random.seed(args.seed); np.random.seed(args.seed) do_crop = args.crop_whitespace and not args.no_crop_whitespace do_clean = args.clean_left_edge and not args.no_clean_left_edge ts = int(args.dataset_size * args.train_ratio); vs = args.dataset_size - ts print("\n" + "="*70) print("SYNTHETIC PARAGRAPH GENERATOR") print("="*70) print(f"Size: {args.dataset_size:,} (train={ts:,}, val={vs:,})") print("\n[1] Loading lines...") ut = load_line_dataset(args.unique_train_dir, "Unique Train") ft = load_line_dataset(args.fixed_train_dir, "Fixed Train") st = load_line_dataset(args.synthetic_train_dir, "Synth Train") uv = load_line_dataset(args.unique_val_dir, "Unique Val") fv = load_line_dataset(args.fixed_val_dir, "Fixed Val") sv = load_line_dataset(args.synthetic_val_dir, "Synth Val") tt = {"unique": len(ut), "fixed": len(ft), "synthetic": len(st)} vt = {"unique": len(uv), "fixed": len(fv), "synthetic": len(sv)} print("\n[2] Verifying isolation...") tp = set(os.path.abspath(p) for p, _ in ut+ft+st) vp = set(os.path.abspath(p) for p, _ in uv+fv+sv) ov = tp & vp print(f" {'WARNING: '+str(len(ov))+' overlap!' if ov else 'Zero overlap confirmed'}") del tp, vp print("\n[3] Merging by writer...") src_t = [("unique", ut)] + ([("fixed", ft)] if ft else []) + ([("synthetic", st)] if st else []) src_v = [("unique", uv)] + ([("fixed", fv)] if fv else []) + ([("synthetic", sv)] if sv else []) tm, _ = merge_lines_by_writer(src_t) vm, _ = merge_lines_by_writer(src_v) print(f" Train: {len(tm)} writers | Val: {len(vm)} writers") tg = SingleWriterParagraphGenerator(tm, tt, args.train_fixed_cap, args.train_synthetic_cap, args.min_lines, args.max_lines, args.max_attempts) vg = SingleWriterParagraphGenerator(vm, vt, args.val_fixed_cap, args.val_synthetic_cap, args.min_lines, args.max_lines, args.max_attempts) td = os.path.join(args.output_dir, "Training") vd = os.path.join(args.output_dir, "Validation") print(f"\n[4] Generating training ({ts:,})...") tc, te, tld = generate_split(tg, ts, td, "Training", args.canvas_width, args.canvas_height, args.padding, args.spacing_min, args.spacing_max, do_crop, do_clean, args.whitespace_threshold, args.edge_pixels, args.output_format, args.gc_interval) print(f"\n[5] Generating validation ({vs:,})...") vc, ve, vld = generate_split(vg, vs, vd, "Validation", args.canvas_width, args.canvas_height, args.padding, args.spacing_min, args.spacing_max, do_crop, do_clean, args.whitespace_threshold, args.edge_pixels, args.output_format, args.gc_interval) print("\n" + "="*70) print("COMPLETE") print("="*70) print(f" Total: {tc+vc:,} (train={tc:,}, val={vc:,}, errors={te+ve:,})") print_stats("Training", tg, tc, tld) print_stats("Validation", vg, vc, vld) print(f"\n Output: {args.output_dir}") print(f" Finished: {datetime.now():%Y-%m-%d %H:%M:%S}") info = os.path.join(args.output_dir, "generation_info.txt") with open(info, "w", encoding="utf-8") as f: f.write(f"Generated: {datetime.now():%Y-%m-%d %H:%M:%S}\n") f.write(f"Size: {args.dataset_size}, Train: {tc}, Val: {vc}\n") f.write(f"Config: {vars(args)}\n") for nm, g in [("Training", tg), ("Validation", vg)]: s = g.get_stats(); f.write(f"\n{nm}:\n") for src in ["unique","fixed","synthetic"]: d = s[src] if d['available'] > 0: f.write(f" {src}: {d['used']:,}/{d['available']:,} ({d['ratio']:.2f}x)\n") print(f" Info: {info}") gc.collect() if __name__ == "__main__": main()