Building Python-Based OSM Validation Rules

Build a tiny, extensible validation framework in Python — a Rule you declare once with a selector, a predicate, a severity, and a message — and a runner that streams a .osm.pbf a single time and yields a typed finding for every element that fails a rule.

Prerequisites

Line these up first; a runner that emits nothing usually means the handler never streamed the file or no rule’s selector matched.

Conceptual minimum

A validation framework earns its keep when adding a new check means writing one small object, not editing a monolith. The design here separates three concerns. A selector decides whether a rule applies to an element at all — usually a tag test such as “is this amenity=fuel?”. A predicate decides whether an applicable element is valid — “does it have a name?”. A finding is what the runner emits when an applicable element fails its predicate, carrying the element’s identity, the rule’s severity, and a human message. Bundling those into a single Rule dataclass makes every check a declarative value you can list, register, and unit-test in isolation, rather than a branch buried in a callback. This is the same separation the editor-time JOSM validation preset draws between a selector and its assertion, expressed in Python instead of MapCSS.

The runner is a pyosmium SimpleHandler. pyosmium streams a PBF element by element in a single pass — nodes, then ways, then relations — invoking a callback per type, so the whole file is never resident in memory. The runner holds a registry: a list of rules grouped by which element types they care about. For each element the stream delivers, the runner dispatches only the rules whose selector accepts that element, runs each predicate, and yields a finding for every failure. Because dispatch is data-driven off the registry, adding a check is appending a Rule to a list; the runner code never changes. That is what makes the framework both extensible and testable — each rule is a pure pair of functions over an element, verifiable without touching a PBF at all. The diagram traces one element through the registry dispatch.

Registry-driven rule dispatch over a streamed OSM element A pyosmium handler streams one element at a time into a rule registry of Rule objects; only rules whose selector accepts the element run their predicate, and each failure yields a finding with element id, severity, and message. pyosmium streams PBF one element rule registry selector · predicate fuel → has name? selector matches highway → surface? selector skips deprecated tag? selector skips run predicate fails no name present yield finding N/123 · warning "fuel has no name" severity + message Selectors that reject the element are skipped; only a failing predicate produces a finding.

Runnable solution

Save as osm_validation.py. It defines the Rule dataclass, a small selector/predicate vocabulary, a registry, and a pyosmium-backed runner that yields findings.

python 
from __future__ import annotations

import logging
from collections.abc import Callable, Iterator
from dataclasses import dataclass, field
from enum import Enum

import osmium

logger = logging.getLogger("osm.validation")

# A read-only view of the element the selector and predicate operate on.
# We snapshot tags and identity so rules stay pure and unit-testable.
@dataclass(frozen=True)
class Element:
    kind: str            # "node" | "way" | "relation"
    id: int
    tags: dict[str, str]
    is_closed: bool = False


class Severity(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"


Selector = Callable[[Element], bool]
Predicate = Callable[[Element], bool]


@dataclass(frozen=True)
class Rule:
    """One pluggable check: applies where `selector` is true, passes where `predicate` is true."""
    code: str
    kinds: frozenset[str]          # which element types this rule inspects
    selector: Selector             # does the rule apply to this element?
    predicate: Predicate           # is an applicable element valid?
    severity: Severity
    message: str


@dataclass(frozen=True)
class Finding:
    code: str
    kind: str
    id: int
    severity: Severity
    message: str


# ---- selector / predicate helpers (composable, testable in isolation) ----
def tag_equals(key: str, value: str) -> Selector:
    return lambda e: e.tags.get(key) == value


def has_tag(key: str) -> Predicate:
    return lambda e: key in e.tags


def lacks_tag(key: str) -> Predicate:
    return lambda e: key not in e.tags


# ---- an example rule registry: append a Rule to extend the framework ----
REGISTRY: list[Rule] = [
    Rule(
        code="fuel-without-name",
        kinds=frozenset({"node", "way"}),
        selector=tag_equals("amenity", "fuel"),
        predicate=has_tag("name"),
        severity=Severity.WARNING,
        message="amenity=fuel has no name",
    ),
    Rule(
        code="highway-without-surface",
        kinds=frozenset({"way"}),
        selector=lambda e: "highway" in e.tags and e.is_closed is False,
        predicate=has_tag("surface"),
        severity=Severity.INFO,
        message="highway way has no surface tag",
    ),
]


class ValidationHandler(osmium.SimpleHandler):
    """Stream a PBF once and collect findings from every registered rule."""

    def __init__(self, rules: list[Rule]) -> None:
        super().__init__()
        self.rules = rules
        self.findings: list[Finding] = []

    def _check(self, el: Element) -> None:
        for rule in self.rules:
            if el.kind not in rule.kinds:
                continue
            if not rule.selector(el):
                continue
            if not rule.predicate(el):
                self.findings.append(
                    Finding(rule.code, el.kind, el.id, rule.severity, rule.message)
                )

    def node(self, n: osmium.osm.Node) -> None:
        self._check(Element("node", n.id, dict(n.tags)))

    def way(self, w: osmium.osm.Way) -> None:
        self._check(Element("way", w.id, dict(w.tags), is_closed=w.is_closed()))

    def relation(self, r: osmium.osm.Relation) -> None:
        self._check(Element("relation", r.id, dict(r.tags)))


def validate(path: str, rules: list[Rule] = REGISTRY) -> Iterator[Finding]:
    """Run every rule over the extract and yield findings."""
    handler = ValidationHandler(rules)
    handler.apply_file(path)
    logger.info("validation complete: %d findings", len(handler.findings))
    yield from handler.findings


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
    for f in validate("extract.osm.pbf"):
        logger.info("%s %s/%d: %s", f.severity.value, f.kind, f.id, f.message)

Step-by-step walkthrough

  1. The Element snapshot. Rules operate on a frozen Element — a copy of the tags plus identity — not on the live pyosmium object. pyosmium reuses its C buffers across callbacks, so snapshotting into a dict keeps rules pure and lets you construct test elements by hand.
  2. Rule as data. The frozen dataclass makes a check a value: code, the element kinds it inspects, a selector, a predicate, a severity, and a message. Nothing about a rule references the runner, so rules are portable and unit-testable.
  3. Selector versus predicate. The split is deliberate: the selector answers “does this rule apply?” and the predicate answers “is the applicable element valid?”. Only an element that the selector accepts and the predicate rejects becomes a finding.
  4. Composable helpers. tag_equals, has_tag, and lacks_tag are small closures that return selectors or predicates, so most rules read declaratively without a bespoke lambda; complex rules can still pass an inline lambda as the highway-without-surface rule does.
  5. The registry. REGISTRY is a plain list. Extending the framework is appending a Rule — the runner never changes, which is the extensibility guarantee. Grouping by kinds lets the dispatcher skip whole rules before evaluating any selector.
  6. Single-pass dispatch. _check runs once per element and iterates the rules, short-circuiting on kinds then selector so most rules cost a dictionary membership test. A failing predicate appends a Finding; a passing one yields nothing.
  7. One stream, all types. apply_file drives the node, way, and relation callbacks in a single sequential pass, so validating a whole extract costs one read of the file regardless of how many rules the registry holds.
  8. validate as a generator. Returning an iterator of findings lets callers stream results into a report, a CSV, or the quarantine discipline from Error Handling in Large OSM Extracts without buffering everything twice.

Verification

Confirm the framework behaves before trusting its findings:

  • Unit-test a rule with no PBF. Build an Element by hand and assert the rule’s outcome, e.g. Element("node", 1, {"amenity": "fuel"}) should produce a fuel-without-name finding while {"amenity": "fuel", "name": "X"} should not.
  • Count against a known extract. Run validate on a small file and compare the finding count for one code to a manual osmium tags-filter count of the same condition.
  • Selector isolation. Confirm that an element the selector rejects yields nothing regardless of the predicate — pass a highway=primary way to the fuel rule and expect zero findings.
  • Severity fidelity. Assert each Finding.severity equals its rule’s declared severity, so downstream filtering by Severity.ERROR is reliable.
  • Single-pass proof. Log element counts in each callback and confirm the file is read once; a second apply_file call would double them.
python 
def test_fuel_without_name():
    rule = next(r for r in REGISTRY if r.code == "fuel-without-name")
    missing = Element("node", 1, {"amenity": "fuel"})
    named = Element("node", 2, {"amenity": "fuel", "name": "Example"})
    assert rule.selector(missing) and not rule.predicate(missing)
    assert rule.selector(named) and rule.predicate(named)

Common errors and fixes

Symptom Root cause One-line fix
Runner yields nothing apply_file never called or wrong path Confirm the path and that validate invokes apply_file.
RuntimeError reading tags after callback Held the live pyosmium object past the callback Snapshot tags with dict(el.tags) into an Element.
Rule never fires Selector too strict or wrong kinds set Loosen the selector and include the right element type.
Every element flagged Predicate and selector logic swapped Selector = applies-to; predicate = is-valid.
Memory climbs on planet file Findings list grows unbounded Stream findings to disk instead of holding all in RAM.
Closed way misclassified is_closed not passed from the way callback Pass w.is_closed() when building the Element.
Duplicate findings per element Same rule registered twice De-duplicate REGISTRY by code at load.

For rules that must also run server-side over a whole database, express the same conditions as an Osmose backend analyser so a scheduled QA run catches what edits miss.

Specification reference

pyosmium exposes OSM data through a SimpleHandler whose node, way, and relation methods are invoked once per element during a single streaming pass of apply_file; element objects and their tag collections are only valid inside the callback, so any data a rule needs must be copied out. See the pyosmium documentation and its handler and reader manual for the handler contract, buffer-lifetime rules, and the location-store options for geometry-aware checks.

Frequently Asked Questions

Why separate the selector from the predicate instead of one function?

The split keeps the two questions a check really asks apart: “does this rule apply to this element?” and “is an applicable element valid?”. Keeping them separate makes findings precise — you only flag elements the rule was meant to judge — and it makes rules composable, since selectors and predicates can be reused across checks. A single combined function tends to conflate “not applicable” with “valid”, which hides real failures.

How do I add a new validation rule?

Append one Rule to the registry list. Give it a code, the element kinds it inspects, a selector that decides applicability, a predicate that decides validity, a severity, and a message. The runner needs no changes because dispatch is driven off the registry. For anything the built-in helpers cannot express, pass an inline lambda as the selector or predicate; it receives the frozen Element and returns a bool.

Why copy tags into a frozen Element rather than use the pyosmium object?

pyosmium reuses its underlying C buffers between callbacks for speed, so the element and tag objects it hands you are only valid inside that single callback. Copying tags into a plain dict on a frozen dataclass decouples rules from that lifetime, prevents a class of use-after-free style bugs, and lets you construct test elements in memory without ever opening a PBF, which is what makes each rule unit-testable.

Can this framework validate geometry, not just tags?

Yes, with one addition. Enable pyosmium’s location store on apply_file so way callbacks can resolve node coordinates, then build the geometry inside the callback and hand it to a geometry-aware predicate. The rule structure is unchanged — selector, predicate, severity, message — only the predicate now inspects a shape instead of tags, which is how self-intersection or unclosed-ring checks slot into the same registry.

Up one level: Authoring OSM Validation Rules.