"""Post-generation validation of exported metadata. Checks that every sample's metadata is internally consistent and safe for downstream ML consumption: the signature was actually detected, its polygon is a valid non-degenerate geometry, every declared bounding box is well-formed and lies on the page, no NaN/Inf values leaked into the JSON, metadata is schema-complete, and the difficulty score is exactly reproducible from its own recorded components and weights. """ from __future__ import annotations import math from dataclasses import dataclass, field from typing import Any from shapely.geometry import Polygon from src.config.schema import DifficultyWeights, ValidationConfig from src.utils.logging_utils import get_logger from src.verification.case_types import CASE_TABLE logger = get_logger(__name__) REQUIRED_TOP_LEVEL_KEYS = ( "document_id", "template", "page", "page_size", "signer", "signature_source", "placement", "rendering", "difficulty", "quality", "verification", "augmentation", "geometry", "metrics", ) # Case types whose ground truth is an intentionally blank target box -- a # real signature is not expected to be detected there. _BLANK_CASE_TYPES = frozenset({"2a", "5a"}) @dataclass class ValidationIssue: field: str message: str severity: str # "error" | "warning" @dataclass class ValidationReport: sample_id: str passed: bool issues: list[ValidationIssue] = field(default_factory=list) @property def errors(self) -> list[ValidationIssue]: return [i for i in self.issues if i.severity == "error"] @property def warnings(self) -> list[ValidationIssue]: return [i for i in self.issues if i.severity == "warning"] class ValidationEngine: """Validates one sample's exported metadata for correctness and completeness. Example: >>> engine = ValidationEngine(cfg.validation, cfg.difficulty) >>> report = engine.validate(metadata) >>> assert report.passed """ def __init__(self, config: ValidationConfig, difficulty_weights: DifficultyWeights) -> None: self.config = config self.weights = difficulty_weights def validate(self, metadata: dict[str, Any]) -> ValidationReport: """Run every validation check against one sample's metadata dict.""" sample_id = str(metadata.get("document_id", "unknown")) issues: list[ValidationIssue] = [] issues += self._check_completeness(metadata) issues += self._check_no_nan(metadata) issues += self._check_signature_exists(metadata) issues += self._check_polygon_valid(metadata) issues += self._check_boxes(metadata) issues += self._check_difficulty_reproducible(metadata) issues += self._check_verification_consistency(metadata) has_errors = any(i.severity == "error" for i in issues) has_warnings = any(i.severity == "warning" for i in issues) passed = not has_errors and not (self.config.strict and has_warnings) return ValidationReport(sample_id=sample_id, passed=passed, issues=issues) def _check_completeness(self, metadata: dict[str, Any]) -> list[ValidationIssue]: issues = [] for key in REQUIRED_TOP_LEVEL_KEYS: if key not in metadata: issues.append(ValidationIssue(key, f"Missing required top-level key '{key}'", "error")) signer = metadata.get("signer", {}) for key in ("name", "role"): if not signer.get(key): issues.append(ValidationIssue(f"signer.{key}", f"Missing signer.{key}", "error")) return issues def _check_no_nan(self, metadata: dict[str, Any], path: str = "") -> list[ValidationIssue]: issues: list[ValidationIssue] = [] # geometry.centroid_px / distance_from_box_center_px are deliberate # NaN sentinels (see GeometryAnalyzer._empty_result) when nothing was # detected -- not a data-quality bug, so they're excluded from the walk. skip_paths: set = set() if not metadata.get("geometry", {}).get("detected", True): skip_paths = {"geometry.centroid_px[0]", "geometry.centroid_px[1]", "geometry.distance_from_box_center_px"} def walk(value: Any, cur_path: str) -> None: if cur_path in skip_paths: return if isinstance(value, dict): for k, v in value.items(): walk(v, f"{cur_path}.{k}" if cur_path else str(k)) elif isinstance(value, (list, tuple)): for i, v in enumerate(value): walk(v, f"{cur_path}[{i}]") elif isinstance(value, float): if math.isnan(value) or math.isinf(value): issues.append(ValidationIssue(cur_path, f"Non-finite value ({value}) at {cur_path}", "error")) walk(metadata, path) return issues @staticmethod def _expects_blank_target(metadata: dict[str, Any]) -> bool: verification = metadata.get("verification") or {} return verification.get("case_type") in _BLANK_CASE_TYPES def _check_signature_exists(self, metadata: dict[str, Any]) -> list[ValidationIssue]: geometry = metadata.get("geometry", {}) if self._expects_blank_target(metadata): return [] if not geometry.get("detected", False): return [ValidationIssue("geometry.detected", "Signature was not detected in the rendered document", "error")] if geometry.get("ink_pixels", 0) <= 0: return [ValidationIssue("geometry.ink_pixels", "Signature detected but has zero ink pixels", "error")] return [] def _check_polygon_valid(self, metadata: dict[str, Any]) -> list[ValidationIssue]: issues: list[ValidationIssue] = [] if not metadata.get("geometry", {}).get("detected", False) and self._expects_blank_target(metadata): return [] poly_pts = metadata.get("geometry", {}).get("signature_polygon_px", []) if len(poly_pts) < 3: issues.append(ValidationIssue("geometry.signature_polygon_px", "Polygon has fewer than 3 points", "error")) return issues try: poly = Polygon(poly_pts) if not poly.is_valid: issues.append(ValidationIssue("geometry.signature_polygon_px", "Polygon geometry is invalid", "error")) elif poly.area <= 0: issues.append(ValidationIssue("geometry.signature_polygon_px", "Polygon has zero area", "error")) except Exception as exc: # pragma: no cover - defensive issues.append(ValidationIssue("geometry.signature_polygon_px", f"Polygon construction failed: {exc}", "error")) return issues def _check_boxes(self, metadata: dict[str, Any]) -> list[ValidationIssue]: issues: list[ValidationIssue] = [] geometry = metadata.get("geometry", {}) page_w, page_h = metadata.get("page_size", [None, None]) eps = self.config.max_coordinate_epsilon for field_name in ("target_box", "actual_box", "signature_bbox_px", "box_px"): box = geometry.get(field_name) if not box or len(box) != 4: issues.append(ValidationIssue(f"geometry.{field_name}", "Box missing or malformed", "error")) continue x1, y1, x2, y2 = box if x2 <= x1 or y2 <= y1: issues.append(ValidationIssue(f"geometry.{field_name}", f"Box has non-positive extent: {box}", "error")) if page_w is not None and page_h is not None: if x1 < -eps or y1 < -eps or x2 > page_w + eps or y2 > page_h + eps: issues.append(ValidationIssue( f"geometry.{field_name}", f"Box {box} lies outside page bounds (0,0,{page_w},{page_h})", "warning", )) return issues def _check_difficulty_reproducible(self, metadata: dict[str, Any]) -> list[ValidationIssue]: difficulty = metadata.get("difficulty", {}) required = ("misplacement", "whitespace", "faintness", "smallness", "low_texture", "score", "tier") if not all(k in difficulty for k in required): return [ValidationIssue("difficulty", "Difficulty block incomplete", "error")] w = self.weights recomputed = ( w.misplacement * difficulty["misplacement"] + w.whitespace * difficulty["whitespace"] + w.faintness * difficulty["faintness"] + w.smallness * difficulty["smallness"] + w.low_texture * difficulty["low_texture"] ) if abs(recomputed - difficulty["score"]) > 1e-3: return [ValidationIssue( "difficulty.score", f"Score {difficulty['score']} does not match recomputed weighted sum {recomputed:.4f}", "error", )] if recomputed < w.easy_threshold: expected_tier = "easy" elif recomputed < w.hard_threshold: expected_tier = "medium" else: expected_tier = "hard" if expected_tier != difficulty["tier"]: return [ValidationIssue( "difficulty.tier", f"Tier '{difficulty['tier']}' does not match expected '{expected_tier}'", "error", )] return [] def _check_verification_consistency(self, metadata: dict[str, Any]) -> list[ValidationIssue]: verification = metadata.get("verification") or {} case_type = verification.get("case_type") if case_type is None: return [] spec = CASE_TABLE.get(case_type) if spec is None: return [ValidationIssue("verification.case_type", f"Unknown case_type '{case_type}'", "error")] issues: list[ValidationIssue] = [] if verification.get("verdict") != spec.verdict: issues.append(ValidationIssue( "verification.verdict", f"verdict '{verification.get('verdict')}' does not match case_type '{case_type}' expected '{spec.verdict}'", "error", )) if verification.get("reject_code") != spec.reject_code: issues.append(ValidationIssue( "verification.reject_code", f"reject_code '{verification.get('reject_code')}' does not match case_type '{case_type}' expected '{spec.reject_code}'", "error", )) return issues @dataclass class ValidationSummary: """Aggregate statistics over a batch of per-sample validation reports.""" total: int passed: int failed: int total_errors: int total_warnings: int failed_sample_ids: list[str] def as_dict(self) -> dict[str, Any]: return { "total": self.total, "passed": self.passed, "failed": self.failed, "pass_rate": round(self.passed / self.total, 4) if self.total else 0.0, "total_errors": self.total_errors, "total_warnings": self.total_warnings, "failed_sample_ids": self.failed_sample_ids, } def summarize(reports: list[ValidationReport]) -> ValidationSummary: """Aggregate a list of per-sample validation reports into a batch summary.""" passed = sum(1 for r in reports if r.passed) return ValidationSummary( total=len(reports), passed=passed, failed=len(reports) - passed, total_errors=sum(len(r.errors) for r in reports), total_warnings=sum(len(r.warnings) for r in reports), failed_sample_ids=[r.sample_id for r in reports if not r.passed], )