Geospatial Compliance Metadata Automation Hub

Aggregating Compliance Metrics Across a Dataset Inventory

To aggregate compliance metrics across a dataset inventory, load the per-dataset check results from SQLite into a pandas DataFrame and reduce them with groupby().mean() — because the pass column is 0 or 1, the mean is the pass rate directly, and a row-wise AND across checks yields an overall compliance score per dataset.

This is the computational core behind Compliance Dashboards for Spatial Catalogs, which covers where these numbers are rendered and gated, and it belongs to the broader Spatial Data Audit Reporting & Compliance Governance program. The script below takes a stored results table and produces the by-check, by-collection, and overall figures a dashboard consumes.

Automated Python Implementation

The script is self-contained: it seeds a small SQLite table so it runs with no external data, then aggregates it three ways. Point --db at your real check_results store to run it against a production catalog.

#!/usr/bin/env python3
"""Aggregate compliance metrics from a SQLite check_results table.

Usage:
    python aggregate_metrics.py --db compliance.db --out metrics.json

With no pre-populated database, --seed inserts a demo run so the script
is runnable standalone.
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import sqlite3

import pandas as pd

SCHEMA = """
CREATE TABLE IF NOT EXISTS check_results (
    dataset_id TEXT NOT NULL,
    collection TEXT,
    check_name TEXT NOT NULL,
    passed     INTEGER NOT NULL,
    run_at     TEXT NOT NULL
);
"""

DEMO_ROWS = [
    ("roads_2025", "transport", "valid_crs", 1),
    ("roads_2025", "transport", "has_license", 1),
    ("roads_2025", "transport", "schema_pass", 1),
    ("parcels_2025", "cadastre", "valid_crs", 1),
    ("parcels_2025", "cadastre", "has_license", 0),
    ("parcels_2025", "cadastre", "schema_pass", 1),
    ("flood_2025", "hazard", "valid_crs", 0),
    ("flood_2025", "hazard", "has_license", 1),
    ("flood_2025", "hazard", "schema_pass", 0),
]


def seed(conn: sqlite3.Connection) -> None:
    """Insert one demo run so the script is runnable with no real data."""
    run_at = dt.datetime.now(dt.timezone.utc).isoformat()
    conn.executescript(SCHEMA)
    conn.executemany(
        "INSERT INTO check_results "
        "(dataset_id, collection, check_name, passed, run_at) "
        "VALUES (?, ?, ?, ?, ?)",
        [(d, c, n, p, run_at) for (d, c, n, p) in DEMO_ROWS],
    )
    conn.commit()


def load_latest(conn: sqlite3.Connection) -> pd.DataFrame:
    """Load the newest run's rows into a DataFrame."""
    latest = conn.execute("SELECT MAX(run_at) FROM check_results").fetchone()[0]
    return pd.read_sql_query(
        "SELECT * FROM check_results WHERE run_at = ?", conn, params=(latest,)
    )


def aggregate(df: pd.DataFrame) -> dict:
    """Compute by-check, by-collection, and overall compliance metrics."""
    by_check = (
        df.groupby("check_name")["passed"].mean().mul(100).round(1)
    ).to_dict()

    by_collection = (
        df.groupby("collection")["passed"].mean().mul(100).round(1)
    ).to_dict()

    # A dataset is fully compliant only if every one of its checks passed.
    wide = df.pivot_table(
        index="dataset_id", columns="check_name",
        values="passed", aggfunc="min",
    )
    fully_compliant = wide.all(axis=1)
    overall = round(float(fully_compliant.mean()) * 100, 1)

    return {
        "run_at": df["run_at"].iloc[0],
        "datasets": int(df["dataset_id"].nunique()),
        "overall_compliant_pct": overall,
        "pass_rate_by_check_pct": by_check,
        "pass_rate_by_collection_pct": by_collection,
        "non_compliant_datasets": sorted(
            fully_compliant[~fully_compliant].index.tolist()
        ),
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Aggregate compliance metrics.")
    parser.add_argument("--db", default=":memory:")
    parser.add_argument("--out", default="metrics.json")
    parser.add_argument("--seed", action="store_true",
                        help="Insert a demo run before aggregating.")
    args = parser.parse_args()

    conn = sqlite3.connect(args.db)
    try:
        conn.executescript(SCHEMA)
        if args.seed or conn.execute(
            "SELECT COUNT(*) FROM check_results"
        ).fetchone()[0] == 0:
            seed(conn)
        metrics = aggregate(load_latest(conn))
    finally:
        conn.close()

    with open(args.out, "w", encoding="utf-8") as fh:
        json.dump(metrics, fh, indent=2)
    print(json.dumps(metrics, indent=2))


if __name__ == "__main__":
    main()

Validation and pipeline integration

Run it standalone against the seeded demo, then confirm the arithmetic:

python aggregate_metrics.py --db :memory: --seed --out metrics.json

# valid_crs passes on 2 of 3 datasets -> expect 66.7
python -c "import json; m=json.load(open('metrics.json')); print(m['pass_rate_by_check_pct']['valid_crs'])"

A pytest fixture pins the aggregation logic so a refactor cannot silently change a published number:

import pandas as pd
from aggregate_metrics import aggregate

def test_overall_requires_all_checks():
    df = pd.DataFrame([
        {"dataset_id": "a", "collection": "x", "check_name": "crs", "passed": 1, "run_at": "t"},
        {"dataset_id": "a", "collection": "x", "check_name": "lic", "passed": 1, "run_at": "t"},
        {"dataset_id": "b", "collection": "x", "check_name": "crs", "passed": 1, "run_at": "t"},
        {"dataset_id": "b", "collection": "x", "check_name": "lic", "passed": 0, "run_at": "t"},
    ])
    m = aggregate(df)
    assert m["overall_compliant_pct"] == 50.0
    assert m["non_compliant_datasets"] == ["b"]

Wire the summary into a scheduled job so metrics refresh on every catalog scan, using the same enforcement discipline as CI/CD validation and policy enforcement for spatial data:

# .github/workflows/metrics.yml
name: Compliance metrics
on:
  schedule:
    - cron: '0 5 * * *'
jobs:
  metrics:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install "pandas>=2.1"
      - run: python aggregate_metrics.py --db compliance.db --out metrics.json
      - uses: actions/upload-artifact@v4
        with:
          name: compliance-metrics
          path: metrics.json

Long-term compliance best practices

  • Aggregate a stored table, never live scans. Compute checks once, persist them, and aggregate from the store so the same numbers reproduce exactly and history is preserved.
  • Define “compliant” as passing every check. A row-wise AND (wide.all(axis=1)) prevents a dataset with one silent failure from being counted as compliant on the strength of its other passing checks.
  • Report the denominator alongside every rate. A pass rate is meaningless without the dataset count; a shrinking inventory can flatter a metric that is actually flat.
  • Group by run_at to expose trends. The same aggregation over multiple runs is a time series — the artifact that proves a remediation sprint worked.
  • Keep the non-compliant list in the output. A headline percentage tells leadership the state; the explicit list of failing dataset_ids tells engineers what to fix.
  • Serialize to JSON, not a rendered view. Emit structured metrics so both the dashboard and a formal report can consume one canonical source.