| |
| """Quality gate — preprocessing.yaml thresholds uygulama. |
| Her kayda quality_gate_pass (bool) + quality_gate_reasons (list) ekler. |
| """ |
| import json, yaml |
| from pathlib import Path |
| from collections import Counter |
|
|
| ROOT = Path("/arf/scratch/stakan/hitit-proje") |
| SOURCES = ROOT / "datasets" / "sources" |
|
|
| def main(): |
| cfg = yaml.safe_load(open(ROOT / "hitit_ocr" / "configs" / "preprocessing.yaml")) |
| gates = cfg['quality_gates'] |
| |
| blur_min = gates['blur_laplacian_min_variance'] |
| exp_min = gates['exposure_mean_min'] |
| exp_max = gates['exposure_mean_max'] |
| cont_min = gates['contrast_std_min'] |
| w_min = gates['resolution_min_w'] |
| h_min = gates['resolution_min_h'] |
| |
| total = 0 |
| passed = 0 |
| reason_counts = Counter() |
| |
| for d in sorted(SOURCES.iterdir()): |
| if not d.is_dir(): continue |
| for mf in ['manifest.jsonl', 'manifest_classification.jsonl', 'manifest_detection.jsonl']: |
| mp = d / mf |
| if not mp.exists(): continue |
| records = [] |
| with open(mp) as f: |
| for line in f: |
| r = json.loads(line) |
| reasons = [] |
| |
| if r.get('integrity_ok') is True: |
| blur = r.get('blur_score') |
| expo = r.get('exposure_mean') |
| cont = r.get('contrast_std') |
| w = r.get('width') |
| h = r.get('height') |
| if blur is not None and blur < blur_min: |
| reasons.append('blur') |
| if expo is not None and (expo < exp_min or expo > exp_max): |
| reasons.append('exposure') |
| if cont is not None and cont < cont_min: |
| reasons.append('contrast') |
| if w and w < w_min: |
| reasons.append('width') |
| if h and h < h_min: |
| reasons.append('height') |
| r['quality_gate_pass'] = len(reasons) == 0 |
| r['quality_gate_reasons'] = reasons |
| if mf == 'manifest.jsonl': |
| total += 1 |
| if r['quality_gate_pass']: passed += 1 |
| for rr in reasons: reason_counts[rr] += 1 |
| else: |
| r['quality_gate_pass'] = None |
| r['quality_gate_reasons'] = None |
| records.append(r) |
| with open(mp, 'w') as f: |
| for r in records: |
| f.write(json.dumps(r, ensure_ascii=False) + '\n') |
| |
| print(f"Total: {total:,}, passed: {passed:,} (%{100*passed/max(1,total):.1f})") |
| print(f"Fail reasons:") |
| for rr, n in reason_counts.most_common(): |
| print(f" {rr}: {n:,}") |
| |
| with open(ROOT / "datasets" / "processed" / "quality_gate_report.json", 'w') as f: |
| json.dump({ |
| "config_version": cfg['version'], |
| "thresholds": gates, |
| "total_scored": total, |
| "passed": passed, |
| "pass_rate": round(100*passed/max(1,total), 2), |
| "fail_reasons": dict(reason_counts), |
| }, f, indent=2, ensure_ascii=False) |
|
|
| if __name__ == '__main__': |
| main() |
|
|
