Geospatial Schema Standardization & Mapping Automation

Generating ISO 19115 Lineage Statements Automatically

A metadata catalogue does not read Parquet. Auditors, INSPIRE portals, and ISO-conformant geoportals expect provenance expressed as ISO 19115-1 LI_Lineage — a human-readable statement plus an ordered list of LI_ProcessStep elements — serialized as ISO 19139 XML. This guide converts the rows produced by Lineage Manifest Generation into exactly that structure using lxml, validates it against the ISO schema, and embeds it in a MD_Metadata record. It is the rendering step that makes an internal audit trail publishable as standards-compliant metadata.

The mapping is mechanical but unforgiving: ISO 19139 is namespace-heavy, element order is schema-enforced, and a single misplaced gco: wrapper makes the whole record invalid against the XSD. Automating it removes the transcription errors that plague hand-authored lineage and guarantees that every published statement traces back to a specific, immutable manifest row. If your source metadata is FGDC rather than born-ISO, the crosswalk in converting FGDC CSDGM to ISO 19115 automatically produces the surrounding record this lineage slots into; both are facets of the Geospatial Schema Architecture & Standards Mapping discipline.

Prerequisites checklist

Step 1: Read and group lineage rows

Load the manifest for a single run and reduce it to the events that belong in published lineage. Not every internal row is a process step — a portal wants meaningful stages (reprojection, attribute casting, validation), not one entry per feature. Group by stage and event_type, order by sequence, and collapse per-feature rows into one process step per logical stage with a count.

python
# iso_lineage/read.py  — python 3.10+, pyarrow >=14
from __future__ import annotations
from dataclasses import dataclass

import pyarrow.dataset as ds
import pyarrow.compute as pc


@dataclass(slots=True)
class ProcessStepInput:
    stage: str
    event_type: str
    outcome: str
    first_time_utc: str      # ISO 8601
    feature_count: int
    tool_version: str | None


def load_process_steps(manifest_root: str, run_id: str) -> list[ProcessStepInput]:
    """Read one run's lineage and reduce it to ordered, published process steps."""
    dataset = ds.dataset(manifest_root, format="parquet", partitioning="hive")
    table = dataset.to_table(filter=ds.field("run_id") == run_id)
    if table.num_rows == 0:
        raise ValueError(f"No lineage rows for run_id={run_id!r}; nothing to publish.")

    # Sort so the earliest event per stage anchors its process-step timestamp.
    order = pc.sort_indices(table, sort_keys=[("sequence", "ascending")])
    table = table.take(order)

    grouped: dict[tuple[str, str], ProcessStepInput] = {}
    for row in table.to_pylist():
        key = (row["stage"], row["event_type"])
        if key not in grouped:
            grouped[key] = ProcessStepInput(
                stage=row["stage"],
                event_type=row["event_type"],
                outcome=row["outcome"],
                first_time_utc=row["event_time_utc"].isoformat(),
                feature_count=0,
                tool_version=row.get("tool_version"),
            )
        grouped[key].feature_count += 1
    return list(grouped.values())

Step 2: Map fields to LI_ProcessStep

Each grouped stage becomes one LI_ProcessStep. The mapping from lineage fields to ISO elements is the contract of this whole guide, so make it explicit:

Lineage field ISO 19115-1 target Notes
stage + event_type LI_ProcessStep/description (gco:CharacterString) Human-readable action, e.g. “CRS reprojection (TRANSFORM)”
first_time_utc LI_ProcessStep/dateTime (gco:DateTime) Earliest event time for the stage; ISO 8601
tool_version LI_ProcessStep/processingInformation rationale PROJ/pyproj build string as the reproducibility anchor
feature_count + outcome LI_ProcessStep/rationale (gco:CharacterString) e.g. “1,204 features, outcome PASS”
stage LI_ProcessStep/processor → CI_ResponsibleParty The software agent that ran the stage

Build each element under the correct namespaces. ISO 19139 places the values inside gco: wrappers, so a bare string is invalid — the value must be <gco:CharacterString>...</gco:CharacterString>.

python
# iso_lineage/build.py  — python 3.10+, lxml >=5.0
from __future__ import annotations
from lxml import etree

from iso_lineage.read import ProcessStepInput

NSMAP = {
    "gmd": "http://www.isotc211.org/2005/gmd",
    "gco": "http://www.isotc211.org/2005/gco",
}


def _gco_char(parent: etree._Element, tag: str, value: str) -> etree._Element:
    """Append <gmd:tag><gco:CharacterString>value</gco:CharacterString></gmd:tag>."""
    el = etree.SubElement(parent, f"{{{NSMAP['gmd']}}}{tag}")
    cs = etree.SubElement(el, f"{{{NSMAP['gco']}}}CharacterString")
    cs.text = value
    return el


def build_process_step(step: ProcessStepInput) -> etree._Element:
    """Render one reduced lineage stage as an ISO 19115-1 LI_ProcessStep."""
    ps = etree.Element(f"{{{NSMAP['gmd']}}}LI_ProcessStep", nsmap=NSMAP)

    description = f"{step.stage} ({step.event_type})"
    _gco_char(ps, "description", description)

    rationale = f"{step.feature_count} features, outcome {step.outcome}"
    _gco_char(ps, "rationale", rationale)

    date_el = etree.SubElement(ps, f"{{{NSMAP['gmd']}}}dateTime")
    dt_el = etree.SubElement(date_el, f"{{{NSMAP['gco']}}}DateTime")
    dt_el.text = step.first_time_utc

    if step.tool_version:
        proc = etree.SubElement(ps, f"{{{NSMAP['gmd']}}}processingInformation")
        _gco_char(proc, "identifier", step.tool_version)

    return ps

Step 3: Build the LI_Lineage tree

Assemble the process steps under a single LI_Lineage element, prefixed by a statement that summarizes provenance in prose. The statement is what a human reads first; the process steps are the machine-auditable detail. Element order matters: statement precedes processStep in the ISO content model.

python
# iso_lineage/build.py (continued)  — python 3.10+, lxml >=5.0
def build_lineage(steps: list[ProcessStepInput], statement: str) -> etree._Element:
    """Assemble an ISO 19115-1 LI_Lineage from ordered process steps."""
    lineage = etree.Element(f"{{{NSMAP['gmd']}}}LI_Lineage", nsmap=NSMAP)

    # statement MUST come before any processStep per the ISO content model.
    _gco_char(lineage, "statement", statement)

    for step in steps:
        wrapper = etree.SubElement(lineage, f"{{{NSMAP['gmd']}}}processStep")
        wrapper.append(build_process_step(step))

    return lineage

Step 4: Validate against the ISO schema

Never publish unvalidated ISO XML — a portal will reject it, and a silently malformed record breaks the audit trail it was meant to prove. Validate against the ISO 19139 XSD with lxml.etree.XMLSchema before the record leaves your process. Because LI_Lineage is a fragment, validate it inside a minimal dataQualityInfo envelope that the schema recognizes, or validate the whole MD_Metadata after embedding (Step 5).

python
# iso_lineage/validate.py  — python 3.10+, lxml >=5.0
from __future__ import annotations
import logging
from pathlib import Path

from lxml import etree

logger = logging.getLogger(__name__)


class LineageValidationError(Exception):
    """Raised when generated lineage XML fails ISO 19139 schema validation."""


def load_schema(xsd_path: Path) -> etree.XMLSchema:
    """Compile the ISO 19139 XSD once; reuse across many records."""
    try:
        return etree.XMLSchema(etree.parse(str(xsd_path)))
    except etree.XMLSchemaParseError as exc:
        logger.critical("Could not compile ISO XSD at %s: %s", xsd_path, exc)
        raise


def validate_document(doc: etree._Element, schema: etree.XMLSchema) -> None:
    """Validate a full MD_Metadata document; raise with the first schema error."""
    if not schema.validate(doc):
        errors = "; ".join(str(e.message) for e in schema.error_log)
        raise LineageValidationError(f"ISO 19139 validation failed: {errors}")
    logger.info("ISO 19139 validation passed.")

Step 5: Embed in the metadata record

Insert the validated LI_Lineage into the record’s MD_Metadata/dataQualityInfo/DQ_DataQuality/lineage slot, then serialize. Placing it under DQ_DataQuality is what ties provenance to the dataset’s declared quality scope — the structure INSPIRE and ISO auditors navigate to.

python
# iso_lineage/embed.py  — python 3.10+, lxml >=5.0
from __future__ import annotations
from pathlib import Path

from lxml import etree

from iso_lineage.build import NSMAP


def embed_lineage(metadata_path: Path, lineage: etree._Element, out_path: Path) -> None:
    """Insert LI_Lineage into DQ_DataQuality/lineage and write the record."""
    parser = etree.XMLParser(remove_blank_text=True)
    tree = etree.parse(str(metadata_path), parser)
    root = tree.getroot()

    dq = root.find(
        ".//gmd:dataQualityInfo/gmd:DQ_DataQuality", namespaces=NSMAP
    )
    if dq is None:
        raise ValueError("MD_Metadata has no DQ_DataQuality element to receive lineage.")

    # Replace any existing lineage; a record carries exactly one.
    for existing in dq.findall("gmd:lineage", namespaces=NSMAP):
        dq.remove(existing)

    wrapper = etree.SubElement(dq, f"{{{NSMAP['gmd']}}}lineage")
    wrapper.append(lineage)

    tree.write(
        str(out_path), pretty_print=True, xml_declaration=True, encoding="UTF-8"
    )

Verification

Confirm the generated record is both well-formed and schema-valid, and that the lineage actually landed where auditors look. Do not trust serialization success alone:

python
# python 3.10+, lxml >=5.0
from pathlib import Path
from lxml import etree
from iso_lineage.read import load_process_steps
from iso_lineage.build import build_lineage, NSMAP
from iso_lineage.validate import load_schema, validate_document
from iso_lineage.embed import embed_lineage

steps = load_process_steps("./_manifest", "run-2026-07-13")
lineage = build_lineage(steps, statement="Reprojected to EPSG:4326 and validated; see process steps.")
embed_lineage(Path("template.xml"), lineage, Path("out.xml"))

doc = etree.parse("out.xml").getroot()
schema = load_schema(Path("schema/gmd/metadataEntity.xsd"))

# 1. The whole record validates against the ISO 19139 XSD.
validate_document(doc, schema)

# 2. Exactly one LI_Lineage exists, under DQ_DataQuality.
found = doc.findall(".//gmd:DQ_DataQuality/gmd:lineage/gmd:LI_Lineage", namespaces=NSMAP)
assert len(found) == 1, f"expected one LI_Lineage, found {len(found)}"

# 3. Every process step carries a dateTime (an auditor's minimum).
for ps in found[0].findall(".//gmd:LI_ProcessStep", namespaces=NSMAP):
    assert ps.find("gmd:dateTime/gco:DateTime", namespaces=NSMAP) is not None, "missing dateTime"

A healthy run logs ISO 19139 validation passed. and produces an out.xml a portal accepts. On the CLI, xmllint --schema schema/gmd/metadataEntity.xsd out.xml --noout prints out.xml validates, and xmllint --xpath 'count(//*[local-name()="LI_ProcessStep"])' out.xml returns the number of published stages.

Troubleshooting

Symptom Likely cause Fix
LineageValidationError naming gco:CharacterString A value was written as bare text instead of inside a gco: wrapper Route every string through _gco_char; ISO 19139 forbids untyped text nodes
element processStep: This element is not expected statement emitted after processStep, violating content order Build statement first (Step 3 ordering); the ISO model is order-sensitive
Namespace prefix gmd not defined on write nsmap not passed to the root element Create the root with nsmap=NSMAP; child elements inherit the declaration
Empty <gmd:lineage/> in output DQ_DataQuality matched but LI_Lineage append targeted the wrong parent Verify the XPath resolves to DQ_DataQuality; embed under a fresh gmd:lineage wrapper
dateTime rejected as invalid Non-ISO 8601 timestamp (missing T or offset) Emit event_time_utc.isoformat() from a tz-aware UTC datetime; never a naive string
XSD compiles but validation always fails offline Schema imports reference remote URLs the resolver cannot fetch Use a locally mirrored XSD bundle with a catalog so all imports resolve on disk

The namespace and ordering failures are the two that dominate. ISO 19139 is a closed content model: every element has a fixed position and a required gco: type wrapper, so treat the XSD as the authority and validate before publish, every time. A record that skips validation will pass your tests and fail at the portal.