| """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", |
| ) |
|
|
| |
| |
| _BLANK_CASE_TYPES = frozenset({"2a", "5a"}) |
|
|
|
|
| @dataclass |
| class ValidationIssue: |
| field: str |
| message: str |
| severity: str |
|
|
|
|
| @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] = [] |
|
|
| |
| |
| |
| 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: |
| 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], |
| ) |
|
|
