# Compliance Dashboards for Spatial Catalogs

A catalog with ten thousand layers can be individually compliant dataset by dataset and still be impossible to govern, because no one can answer the questions leadership and auditors actually ask: what fraction of the catalog carries a valid CRS, how many datasets have a resolvable license, is schema conformance trending up or down. A compliance dashboard turns thousands of per-dataset check results into a handful of catalog-wide metrics with history behind them. This guide, part of Spatial Data Audit Reporting & Compliance Governance, builds that pipeline: run per-dataset checks, persist the outcomes in SQLite, aggregate them with pandas, and render a static HTML dashboard that regenerates deterministically on every scan.

# Prerequisites

1. Python 3.11+ in an isolated virtual environment (python -m venv .venv)
2. pandas>=2.1 — aggregation and pivoting of per-dataset results
3. jinja2>=3.1 — templating the static HTML dashboard
4. geopandas>=0.14 and pyproj>=3.6 — the CRS/license/schema checks that feed the dashboard
5. jsonschema>=4.21 — schema-conformance checks against a catalog profile
6. Python’s standard-library sqlite3 (no install needed) for the results store
7. Environment variable CATALOG_DB pointing at the SQLite results file, e.g. ./compliance.db
8. A dataset inventory (CSV, catalog export, or STAC) listing every layer’s id, path, and collection

Install the stack:

pip install "pandas>=2.1" "jinja2>=3.1" "geopandas>=0.14" \
    "pyproj>=3.6" "jsonschema>=4.21"

# Concept & Spec Reference

A compliance dashboard is a deterministic projection of a per-dataset results table. The unit of measurement is the check: a named boolean predicate applied to one dataset, such as “has a valid, resolvable CRS.” Each check produces one row; the dashboard aggregates rows into pass rates. Keeping checks atomic and independently named is what lets you pivot the same data by obligation, by collection, and over time. The individual predicates reuse validators you have already built — CRS resolution via pyproj, license presence from your metadata schema validation and linting rules, and geometry integrity from automated topology and geometry validation.

# Metric definitions

# Results table schema

# Implementation Walkthrough

# Step 1 — Define atomic, named checks

Each check is a small function returning (passed, detail). Keeping them independent means a new obligation is a new function, not a rewrite. The rationale: atomic checks are individually testable and their results pivot cleanly.

import pyproj

def check_valid_crs(record: dict) -> tuple[bool, str]:
    """Pass if the declared CRS resolves to a known authority code."""
    crs_value = record.get("crs")
    if not crs_value:
        return False, "no CRS declared"
    try:
        crs = pyproj.CRS.from_user_input(crs_value)
    except pyproj.exceptions.CRSError as exc:
        return False, f"unresolvable CRS: {exc}"
    return bool(crs.to_authority()), (
        "ok" if crs.to_authority() else "CRS has no authority code"
    )

def check_has_license(record: dict) -> tuple[bool, str]:
    """Pass if a non-empty license identifier is present."""
    lic = (record.get("license") or "").strip()
    return (bool(lic), "ok" if lic else "missing license")

# Step 2 — Persist per-dataset results in SQLite

Write outcomes to a durable table so the dashboard becomes a historical record. The rationale: recomputation gives you a snapshot; persistence gives you a trend and an auditable point-in-time answer.

import sqlite3
import datetime as dt

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

CHECKS = {
    "valid_crs": check_valid_crs,
    "has_license": check_has_license,
}

def run_and_store(db_path: str, inventory: list[dict]) -> str:
    """Run all checks over the inventory and persist results. Returns run_at."""
    run_at = dt.datetime.now(dt.timezone.utc).isoformat()
    conn = sqlite3.connect(db_path)
    try:
        conn.executescript(SCHEMA)
        rows = []
        for record in inventory:
            for name, fn in CHECKS.items():
                passed, detail = fn(record)
                rows.append((
                    record["dataset_id"], record.get("collection"),
                    name, int(passed), detail, run_at,
                ))
        conn.executemany(
            "INSERT INTO check_results "
            "(dataset_id, collection, check_name, passed, detail, run_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            rows,
        )
        conn.commit()
    finally:
        conn.close()
    return run_at

# Step 3 — Aggregate to catalog metrics with pandas

Load the latest run into a DataFrame and pivot. The rationale: a single grouped aggregation yields both the by-check and by-collection views the dashboard needs.

import pandas as pd
import sqlite3

def aggregate_latest(db_path: str) -> dict:
    """Return pass rates by check and by collection for the newest run."""
    conn = sqlite3.connect(db_path)
    try:
        latest = conn.execute(
            "SELECT MAX(run_at) FROM check_results"
        ).fetchone()[0]
        df = pd.read_sql_query(
            "SELECT * FROM check_results WHERE run_at = ?",
            conn, params=(latest,),
        )
    finally:
        conn.close()

    by_check = (
        df.groupby("check_name")["passed"]
        .mean().mul(100).round(1)
        .rename("pass_rate_pct").reset_index()
    )
    by_collection = (
        df.groupby("collection")["passed"]
        .mean().mul(100).round(1)
        .rename("pass_rate_pct").reset_index()
    )
    return {
        "run_at": latest,
        "datasets": int(df["dataset_id"].nunique()),
        "by_check": by_check.to_dict("records"),
        "by_collection": by_collection.to_dict("records"),
    }

# Step 4 — Render a static dashboard

Template the metrics into self-contained HTML. The rationale: a static file is archivable, diffable, and needs no server, so it can be attached to an audit package unchanged.

from jinja2 import Environment, BaseLoader

DASHBOARD_TMPL = """<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<title>Catalog Compliance — {{ run_at }}</title></head>
<body>
<h1>Catalog Compliance Dashboard</h1>
<p>{{ datasets }} datasets evaluated at {{ run_at }}.</p>
<h2>By check</h2>
<table><tr><th>Check</th><th>Pass rate</th></tr>
{% for row in by_check %}<tr><td>{{ row.check_name }}</td>
<td>{{ row.pass_rate_pct }}%</td></tr>{% endfor %}
</table>
<h2>By collection</h2>
<table><tr><th>Collection</th><th>Pass rate</th></tr>
{% for row in by_collection %}<tr><td>{{ row.collection }}</td>
<td>{{ row.pass_rate_pct }}%</td></tr>{% endfor %}
</table>
</body></html>
"""

def render_dashboard(metrics: dict, out_path: str) -> None:
    """Write a self-contained HTML dashboard from aggregated metrics."""
    env = Environment(loader=BaseLoader(), autoescape=True)
    html = env.from_string(DASHBOARD_TMPL).render(**metrics)
    with open(out_path, "w", encoding="utf-8") as fh:
        fh.write(html)

# Validation & CI Integration

# Guard against regressions with a threshold gate

A dashboard is descriptive; a gate is prescriptive. Fail CI when any catalog metric drops below its policy floor so compliance cannot silently erode between releases.

import sys

THRESHOLDS = {"valid_crs": 95.0, "has_license": 90.0, "schema_pass": 98.0}

def enforce_thresholds(metrics: dict) -> int:
    """Exit non-zero if any check's pass rate is below its threshold."""
    failures = []
    rates = {r["check_name"]: r["pass_rate_pct"] for r in metrics["by_check"]}
    for check, floor in THRESHOLDS.items():
        actual = rates.get(check, 0.0)
        if actual < floor:
            failures.append(f"{check}: {actual}% < {floor}% floor")
    for msg in failures:
        print(f"THRESHOLD BREACH: {msg}")
    return 1 if failures else 0

# GitHub Actions workflow

# .github/workflows/compliance-dashboard.yml
name: Catalog compliance dashboard
on:
  schedule:
    - cron: '0 6 * * 1'
  workflow_dispatch:
jobs:
  dashboard:
    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" "jinja2>=3.1" "pyproj>=3.6"
      - run: python scripts/build_dashboard.py --db compliance.db --out public/compliance.html
      - uses: actions/upload-artifact@v4
        with:
          name: compliance-dashboard
          path: public/compliance.html

# Pytest coverage

import pandas as pd
from scripts.build_dashboard import aggregate_latest, enforce_thresholds

def test_thresholds_pass_when_rates_high():
    metrics = {"by_check": [
        {"check_name": "valid_crs", "pass_rate_pct": 99.0},
        {"check_name": "has_license", "pass_rate_pct": 92.0},
        {"check_name": "schema_pass", "pass_rate_pct": 99.5},
    ]}
    assert enforce_thresholds(metrics) == 0

def test_thresholds_fail_on_low_crs():
    metrics = {"by_check": [{"check_name": "valid_crs", "pass_rate_pct": 80.0}]}
    assert enforce_thresholds(metrics) == 1

# Derivative & Lineage Management

A dashboard metric is only as honest as the check behind it, and checks drift as datasets are transformed. When a pipeline reprojects, clips, or joins a layer, re-run its checks and write fresh rows rather than editing old ones — the history of a dataset’s compliance state is itself audit evidence. Tie the has_lineage metric to the spatial data lineage and provenance tracking graphs so a derived layer with no provenance is visibly counted as non-compliant. When a metric improves after a remediation sprint, the timestamped rows in SQLite are the proof; export the underlying results into a formal automated compliance report when an auditor needs the detail behind a headline number.

# Pitfalls & Resolution

# Related

• Spatial Data Audit Reporting & Compliance Governance — the parent guide positioning dashboards within the audit program
• Aggregating Compliance Metrics Across a Dataset Inventory — a self-contained pandas/SQLite script for the aggregation core
• Spatial Data Lineage & Provenance Tracking — supplies the has_lineage signal the dashboard reports
• Automated Compliance Report Generation — turn dashboard metrics into formal audit deliverables
• Metadata Schema Validation & Linting — the validators behind the license and schema checks