Exporting Planning Authority Submission Packages #

# requirements.txt
# geopandas==1.0.1
# psycopg==3.2.1
# fpdf2==2.7.9
# pyproj==3.6.1

import csv
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path

import geopandas as gpd
from fpdf import FPDF

TARGET_EPSG = 27700  # British National Grid; substitute your site's EPSG

# Canonical attribute schema every submission layer must expose.
FEATURE_COLUMNS = ["feature_id", "context", "feature_type", "rule", "list_entry",
                   "monument", "distance_m", "geom"]


def export_package(conn_str: str, run_id: str, site_code: str, out_root: str) -> Path:
    """Assemble a GeoPackage + CSV + PDF + metadata sidecar for one validation run."""
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d")
    pkg = Path(out_root) / f"{site_code}_submission_{stamp}"
    pkg.mkdir(parents=True, exist_ok=True)

    # 1. Pull features joined to their violation rows for this run.
    features = gpd.read_postgis(
        """
        SELECT f.feature_id, f.context, f.feature_type,
               v.rule, v.list_entry, v.monument, v.distance_m, f.geom
        FROM   features f
        JOIN   compliance.boundary_violations v ON v.feature_id = f.feature_id
        WHERE  v.run_id = %(run_id)s
        """,
        conn_str, geom_col="geom", params={"run_id": run_id},
    ).to_crs(epsg=TARGET_EPSG)

    # 2. Pull only the designations actually implicated by this run.
    designations = gpd.read_postgis(
        """
        SELECT d.list_entry, d.designation_type, d.name, d.grade, d.geom
        FROM   compliance.designations d
        WHERE  d.list_entry IN (
            SELECT DISTINCT list_entry FROM compliance.boundary_violations
            WHERE run_id = %(run_id)s AND list_entry IS NOT NULL)
        """,
        conn_str, geom_col="geom", params={"run_id": run_id},
    ).to_crs(epsg=TARGET_EPSG)

    # 3. GeoPackage — the authoritative geometry, two layers in one file.
    gpkg = pkg / f"{site_code}_submission.gpkg"
    features.to_file(gpkg, layer="affected_features", driver="GPKG")
    designations.to_file(gpkg, layer="applicable_designations", driver="GPKG")

    # 4. CSV index — tabular, no geometry, for spreadsheet review.
    csv_path = pkg / f"{site_code}_feature_index.csv"
    with csv_path.open("w", newline="") as fh:
        writer = csv.writer(fh)
        writer.writerow(["feature_id", "context", "feature_type",
                         "rule", "list_entry", "monument", "distance_m"])
        for _, r in features.drop(columns="geom").iterrows():
            writer.writerow([r.feature_id, r.context, r.feature_type,
                             r.rule, r.list_entry, r.monument, r.distance_m])

    # 5. PDF cover report — human summary of the run.
    pdf_path = pkg / f"{site_code}_report.pdf"
    _write_pdf(pdf_path, site_code, run_id, features, designations)

    # 6. Metadata sidecar (ISO 19115 aligned) + manifest with checksums.
    _write_metadata(pkg / "metadata.json", site_code, run_id, len(features))
    _write_manifest(pkg)
    return pkg


def _write_pdf(path, site_code, run_id, features, designations):
    pdf = FPDF()
    pdf.add_page()
    pdf.set_font("Helvetica", "B", 16)
    pdf.cell(0, 10, f"Heritage Constraint Submission - {site_code}", ln=True)
    pdf.set_font("Helvetica", "", 10)
    pdf.cell(0, 7, f"Validation run: {run_id}", ln=True)
    pdf.cell(0, 7, f"Generated: {datetime.now(timezone.utc).isoformat(timespec='seconds')}", ln=True)
    pdf.cell(0, 7, f"CRS: EPSG:{TARGET_EPSG} (British National Grid)", ln=True)
    pdf.ln(4)
    pdf.set_font("Helvetica", "B", 12)
    pdf.cell(0, 8, "Summary", ln=True)
    pdf.set_font("Helvetica", "", 10)
    inside = (features["rule"] == "inside_scheduled").sum()
    near = (features["rule"] == "within_buffer").sum()
    pdf.cell(0, 6, f"Features within scheduled areas: {inside}", ln=True)
    pdf.cell(0, 6, f"Features within consultation buffer: {near}", ln=True)
    pdf.cell(0, 6, f"Distinct designations implicated: {designations['list_entry'].nunique()}", ln=True)
    pdf.output(str(path))


def _write_metadata(path, site_code, run_id, feature_count):
    meta = {
        "title": f"Heritage constraint submission for {site_code}",
        "run_id": run_id,
        "reference_system": f"EPSG:{TARGET_EPSG}",
        "date_stamp": datetime.now(timezone.utc).date().isoformat(),
        "lineage": "Features validated against Historic England NHLE scheduled monuments.",
        "feature_count": feature_count,
        "standard": "ISO 19115",
    }
    path.write_text(json.dumps(meta, indent=2))


def _write_manifest(pkg: Path):
    """SHA-256 every file so the recipient can verify integrity."""
    entries = []
    for f in sorted(pkg.iterdir()):
        if f.name == "manifest.json":
            continue
        digest = hashlib.sha256(f.read_bytes()).hexdigest()
        entries.append({"file": f.name, "bytes": f.stat().st_size, "sha256": digest})
    (pkg / "manifest.json").write_text(json.dumps({"files": entries}, indent=2))

# GDAL 3.9.1 / ogr2ogr
ogr2ogr -f GPKG SGC24_submission.gpkg \
  PG:"dbname=archaeology_db user=heritage" \
  -sql "SELECT feature_id, context, geom FROM features WHERE site_code='SGC24'" \
  -nln affected_features -t_srs EPSG:27700

from osgeo import ogr

def verify_package(pkg):
    manifest = json.loads((pkg / "manifest.json").read_text())
    for entry in manifest["files"]:
        digest = hashlib.sha256((pkg / entry["file"]).read_bytes()).hexdigest()
        assert digest == entry["sha256"], f"checksum drift: {entry['file']}"
    ds = ogr.Open(str(next(pkg.glob("*.gpkg"))))
    layers = {ds.GetLayer(i).GetName() for i in range(ds.GetLayerCount())}
    assert {"affected_features", "applicable_designations"} <= layers
    srs = ds.GetLayerByName("affected_features").GetSpatialRef()
    assert srs.GetAuthorityCode(None) == "27700"
    print("package verified:", pkg.name)