How to Validate GeoPackage OGC Compliance

import sqlite3
import subprocess
import sys
import json
from pathlib import Path

def validate_geopackage(gpkg_path: str) -> dict:
    """Validate OGC GeoPackage compliance using SQLite schema checks + GDAL."""
    path = Path(gpkg_path)
    if not path.exists():
        return {"status": "FAIL", "error": "File not found"}

    report = {"file": str(path), "compliant": True, "checks": []}

    try:
        with sqlite3.connect(str(path)) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()

            # 1. Mandatory OGC Table Presence
            mandatory_tables = {
                "gpkg_spatial_ref_sys", "gpkg_contents", 
                "gpkg_geometry_columns", "gpkg_extensions"
            }
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
            existing = {row["name"] for row in cursor.fetchall()}
            missing = mandatory_tables - existing
            if missing:
                report["compliant"] = False
                report["checks"].append({"check": "mandatory_tables", "status": "FAIL", "missing": sorted(missing)})
            else:
                report["checks"].append({"check": "mandatory_tables", "status": "PASS"})

            # 2. gpkg_contents Validation
            cursor.execute("SELECT table_name, data_type, srs_id FROM gpkg_contents;")
            contents = cursor.fetchall()
            if not contents:
                report["compliant"] = False
                report["checks"].append({"check": "gpkg_contents", "status": "FAIL", "message": "Empty contents table"})
            else:
                valid_types = {"features", "attributes", "tiles"}
                invalid = [r["data_type"] for r in contents if r["data_type"] not in valid_types]
                if invalid:
                    report["compliant"] = False
                    report["checks"].append({"check": "gpkg_contents", "status": "FAIL", "invalid_types": invalid})
                else:
                    report["checks"].append({"check": "gpkg_contents", "status": "PASS"})

            # 3. Spatial Reference Integrity
            cursor.execute("SELECT srs_id, organization, organization_coordsys_id FROM gpkg_spatial_ref_sys WHERE srs_id != 0;")
            srs_rows = cursor.fetchall()
            if not srs_rows:
                report["checks"].append({"check": "spatial_ref_sys", "status": "WARN", "message": "Only undefined SRS (0) found"})
            else:
                report["checks"].append({"check": "spatial_ref_sys", "status": "PASS", "count": len(srs_rows)})

    except sqlite3.Error as e:
        report["compliant"] = False
        report["checks"].append({"check": "sqlite_connection", "status": "FAIL", "error": str(e)})
        return report

    # 4. GDAL Geometry & WKB Validation (ogrinfo)
    try:
        result = subprocess.run(
            ["ogrinfo", "-so", "-al", str(path)],
            capture_output=True, text=True, check=True, timeout=30
        )
        if "ERROR" in result.stderr.upper():
            report["compliant"] = False
            report["checks"].append({"check": "gdal_geometry", "status": "FAIL", "details": result.stderr.strip()})
        else:
            report["checks"].append({"check": "gdal_geometry", "status": "PASS"})
    except FileNotFoundError:
        report["checks"].append({"check": "gdal_geometry", "status": "SKIP", "message": "GDAL/ogrinfo not installed"})
    except subprocess.TimeoutExpired:
        report["checks"].append({"check": "gdal_geometry", "status": "FAIL", "message": "Validation timed out"})
    except subprocess.CalledProcessError as e:
        report["checks"].append({"check": "gdal_geometry", "status": "FAIL", "details": e.stderr.strip()})

    return report

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python validate_gpkg.py <path_to_gpkg>")
        sys.exit(1)
    print(json.dumps(validate_geopackage(sys.argv[1]), indent=2))