Rule-Based Styling Engines for Automated Cartographic Workflows
Applying symbology by hand in a GUI works for a single map, but it breaks the moment you need to reproduce that output at a different scale, swap the underlying dataset, or hand the workflow to a colleague. A rule-based styling engine replaces per-feature manual edits with a declarative logic layer: feature attributes are evaluated against conditional expressions, and rendering properties — fill colour, stroke weight, opacity, label size — are assigned deterministically. The result is styling logic that lives in version control, passes through CI/CD, and produces identical output whether it runs on a developer laptop or a headless cloud worker.
This page covers the full engineering path: normalizing input data so rules evaluate correctly, defining a safe and auditable rule schema, building the evaluation pipeline, wiring in label collision avoidance during rendering, and synchronizing the engine with dynamic legend generation on export. The implementation runs on Python 3.11 with geopandas 0.14, shapely 2.0, simpleeval 0.9.13, and PyYAML 6.0.
Prerequisites and Environment Configuration
python >= 3.11
geopandas == 0.14.4
shapely == 2.0.6
simpleeval == 0.9.13
PyYAML == 6.0.2
jsonschema == 4.23.0
pytest == 8.3.2
pytest-snapshot== 0.9.0
Install with:
pip install geopandas==0.14.4 shapely==2.0.6 simpleeval==0.9.13 \
pyyaml==6.0.2 jsonschema==4.23.0 pytest==8.3.2 pytest-snapshot==0.9.0Input data must be in EPSG:4326 (WGS 84) or a well-defined projected CRS that you declare explicitly. Mixed-CRS inputs — GeoJSON in 4326 merged with a Shapefile in a local projected system — cause silent coordinate mismatches that produce geometrically valid but cartographically wrong output. Reproject everything to a single canonical CRS before the normalization step using gdf.to_crs(epsg=4326).
Conceptual Foundation: Priority-Order Rule Evaluation
A styling engine is, at its core, a sorted filter chain. Each rule carries three pieces of state: a condition (a boolean expression over feature attributes), a style (a dict of rendering properties), and a priority (an integer establishing override order). The engine iterates every feature through every rule in ascending priority order, accumulating style assignments. Because sorting is stable and priority integers are enforced unique at load time, the output is deterministic: the same input always produces the same styled GeoDataFrame.
The key algorithmic choice is evaluation strategy: evaluate all rules for every feature (full-pass), or stop at the first match (short-circuit). Full-pass evaluation is correct for layered overrides — a base rule sets a default grey fill, and a higher-priority rule overrides only the stroke colour for motorways — but it is O(F × R) where F is feature count and R is rule count. Short-circuit evaluation is O(F × R/2) on average but only correct when rules are mutually exclusive. Choose full-pass for cartographic workflows where partial overrides are common; short-circuit for classification schemes with discrete, non-overlapping categories.
Safe expression parsing is non-negotiable. Python’s eval() executes arbitrary code — a malformed rule file or an upstream data injection can corrupt the entire pipeline. simpleeval’s EvalWithCompoundTypes evaluates arithmetic and comparison expressions against an explicit names dict, rejects attribute access on unknown names, and raises FeatureNotAvailable for any identifier not in the dict. This gives you sandboxed evaluation with no external process overhead.
Step-by-Step Implementation
Step 1 — Normalize Attribute Data
Rule conditions compare strings exactly. Before evaluation, cast every categorical field to lowercase and strip whitespace. Cast numeric fields that may arrive as strings (common in GeoJSON properties exported from web services) to float. Fill nulls with a sentinel value your rules can match explicitly.
import geopandas as gpd
import pandas as pd
def normalize_attributes(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
"""
Normalize a GeoDataFrame so rule conditions evaluate deterministically.
Returns a copy — never mutates the input.
"""
gdf = gdf.copy()
# Categorical fields: lowercase + strip
cat_cols = gdf.select_dtypes(include="object").columns
for col in cat_cols:
gdf[col] = gdf[col].str.strip().str.lower().fillna("__null__")
# Numeric fields that arrived as strings
for col in gdf.columns:
if col == gdf.geometry.name:
continue
try:
gdf[col] = pd.to_numeric(gdf[col], errors="ignore")
except TypeError:
pass
# Drop null geometries — they will crash every renderer
before = len(gdf)
gdf = gdf[gdf.geometry.notna() & gdf.geometry.is_valid].reset_index(drop=True)
dropped = before - len(gdf)
if dropped:
import warnings
warnings.warn(f"normalize_attributes: dropped {dropped} null/invalid geometries")
return gdfValidate join cardinality before any spatial merge. A one-to-many join duplicates geometries and corrupts rendering order; detect it early:
def assert_join_cardinality(left: gpd.GeoDataFrame, right: pd.DataFrame, key: str) -> None:
dupes = right[right.duplicated(subset=key)]
if not dupes.empty:
raise ValueError(
f"Right table has {len(dupes)} duplicate '{key}' values — "
"merge would produce one-to-many geometry duplication."
)Step 2 — Define the Rule Schema in YAML
Keep rules in YAML for human readability; validate them against a JSON Schema at load time so malformed rules fail fast rather than silently applying wrong styles.
# rules/land_use.yaml
rules:
- id: residential_fill
condition: "land_use == 'residential'"
style:
fill: "#F0EAD6"
stroke: "#C8B89A"
stroke_width: 0.5
opacity: 1.0
priority: 10
fallback: false
- id: commercial_fill
condition: "land_use == 'commercial'"
style:
fill: "#E8D5B7"
stroke: "#C8A882"
stroke_width: 0.5
opacity: 1.0
priority: 10
fallback: false
- id: motorway_stroke_override
condition: "road_class == 'motorway' and lane_count >= 4"
style:
stroke: "#CC3300"
stroke_width: 2.5
priority: 80
fallback: false
- id: default_fallback
condition: "True"
style:
fill: "#E8E8E8"
stroke: "#AAAAAA"
stroke_width: 0.3
opacity: 0.8
priority: 1
fallback: trueThe JSON Schema that validates this structure:
RULE_SCHEMA = {
"type": "object",
"required": ["rules"],
"properties": {
"rules": {
"type": "array",
"items": {
"type": "object",
"required": ["id", "condition", "style", "priority"],
"properties": {
"id": {"type": "string"},
"condition": {"type": "string"},
"style": {"type": "object"},
"priority": {"type": "integer", "minimum": 0},
"fallback": {"type": "boolean"}
},
"additionalProperties": False
}
}
}
}Step 3 — Load and Validate Rules
import yaml
import jsonschema
from pathlib import Path
def load_rules(rules_path: Path) -> list[dict]:
"""Load and validate rules from a YAML file."""
with rules_path.open() as f:
doc = yaml.safe_load(f)
jsonschema.validate(doc, RULE_SCHEMA)
rules = doc["rules"]
# Enforce unique priority integers — ties produce non-deterministic output
priorities = [r["priority"] for r in rules]
if len(priorities) != len(set(priorities)):
from collections import Counter
dupes = [p for p, n in Counter(priorities).items() if n > 1]
raise ValueError(f"Duplicate priority values detected: {dupes}")
# Sort ascending: lower priority evaluated first, higher priority wins last
return sorted(rules, key=lambda r: r["priority"])Step 4 — Safe Condition Evaluation
simpleeval’s EvalWithCompoundTypes exposes only the feature’s attribute dict as names, blocking any access to builtins, imports, or OS functions.
from simpleeval import EvalWithCompoundTypes, FeatureNotAvailable
_EVALUATOR = EvalWithCompoundTypes()
def evaluate_condition(condition_str: str, feature_attrs: dict) -> bool:
"""
Safely evaluate a condition string against a feature's attribute dict.
Returns False (not True) on evaluation errors so the pipeline continues.
"""
try:
result = _EVALUATOR.eval(condition_str, names=feature_attrs)
return bool(result)
except FeatureNotAvailable as exc:
import warnings
warnings.warn(f"Unknown attribute in condition '{condition_str}': {exc}")
return False
except Exception as exc:
warnings.warn(f"Condition evaluation error '{condition_str}': {exc}")
return FalseStep 5 — Run the Evaluation Pipeline
Full-pass evaluation: iterate all rules for every feature, accumulating style overrides in priority order.
def apply_rules(
gdf: gpd.GeoDataFrame,
rules: list[dict],
default_style: dict | None = None,
) -> gpd.GeoDataFrame:
"""
Apply sorted rules to every feature in gdf.
Returns gdf with a new 'resolved_style' column (list of dicts).
"""
if default_style is None:
default_style = {
"fill": "#CCCCCC",
"stroke": "#888888",
"stroke_width": 0.5,
"opacity": 1.0,
}
resolved_styles: list[dict] = []
for _, row in gdf.iterrows():
attrs = {k: v for k, v in row.items() if k != gdf.geometry.name}
accumulated = dict(default_style) # start from global fallback
for rule in rules: # already sorted ascending by priority
if evaluate_condition(rule["condition"], attrs):
accumulated.update(rule["style"]) # higher priority overrides
resolved_styles.append(accumulated)
gdf = gdf.copy()
gdf["resolved_style"] = resolved_styles
return gdfComplete Working Code Example
"""
rule_engine.py — end-to-end rule-based styling engine for GeoDataFrame inputs.
Usage:
from rule_engine import build_styled_gdf
styled = build_styled_gdf("data/land_use.gpkg", "rules/land_use.yaml")
"""
import warnings
from pathlib import Path
import geopandas as gpd
import pandas as pd
import yaml
import jsonschema
from simpleeval import EvalWithCompoundTypes, FeatureNotAvailable
# ─── Schema ───────────────────────────────────────────────────────────────────
RULE_SCHEMA = {
"type": "object",
"required": ["rules"],
"properties": {
"rules": {
"type": "array",
"items": {
"type": "object",
"required": ["id", "condition", "style", "priority"],
"properties": {
"id": {"type": "string"},
"condition": {"type": "string"},
"style": {"type": "object"},
"priority": {"type": "integer", "minimum": 0},
"fallback": {"type": "boolean"},
},
"additionalProperties": False,
},
}
},
}
_EVALUATOR = EvalWithCompoundTypes()
# ─── Data preparation ─────────────────────────────────────────────────────────
def normalize_attributes(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
gdf = gdf.copy()
cat_cols = gdf.select_dtypes(include="object").columns
for col in cat_cols:
gdf[col] = gdf[col].str.strip().str.lower().fillna("__null__")
for col in gdf.columns:
if col == gdf.geometry.name:
continue
try:
gdf[col] = pd.to_numeric(gdf[col], errors="ignore")
except TypeError:
pass
before = len(gdf)
gdf = gdf[gdf.geometry.notna() & gdf.geometry.is_valid].reset_index(drop=True)
if (dropped := before - len(gdf)):
warnings.warn(f"Dropped {dropped} null/invalid geometries")
return gdf
# ─── Rule loading ─────────────────────────────────────────────────────────────
def load_rules(rules_path: Path) -> list[dict]:
with rules_path.open() as f:
doc = yaml.safe_load(f)
jsonschema.validate(doc, RULE_SCHEMA)
rules = doc["rules"]
from collections import Counter
dupes = [p for p, n in Counter(r["priority"] for r in rules).items() if n > 1]
if dupes:
raise ValueError(f"Duplicate rule priorities: {dupes}")
return sorted(rules, key=lambda r: r["priority"])
# ─── Condition evaluation ──────────────────────────────────────────────────────
def evaluate_condition(condition_str: str, attrs: dict) -> bool:
try:
return bool(_EVALUATOR.eval(condition_str, names=attrs))
except FeatureNotAvailable as exc:
warnings.warn(f"Unknown attribute in '{condition_str}': {exc}")
return False
except Exception as exc:
warnings.warn(f"Eval error '{condition_str}': {exc}")
return False
# ─── Evaluation pipeline ───────────────────────────────────────────────────────
_DEFAULT_STYLE = {
"fill": "#E8E8E8",
"stroke": "#AAAAAA",
"stroke_width": 0.5,
"opacity": 1.0,
}
def apply_rules(gdf: gpd.GeoDataFrame, rules: list[dict]) -> gpd.GeoDataFrame:
resolved: list[dict] = []
for _, row in gdf.iterrows():
attrs = {k: v for k, v in row.items() if k != gdf.geometry.name}
accumulated = dict(_DEFAULT_STYLE)
for rule in rules:
if evaluate_condition(rule["condition"], attrs):
accumulated.update(rule["style"])
resolved.append(accumulated)
gdf = gdf.copy()
gdf["resolved_style"] = resolved
return gdf
# ─── Legend extraction ────────────────────────────────────────────────────────
def extract_legend_entries(rules: list[dict]) -> list[dict]:
"""Return one legend entry per non-fallback rule, for dynamic legend generation."""
return [
{"label": r["id"].replace("_", " ").title(), "style": r["style"]}
for r in rules
if not r.get("fallback", False)
]
# ─── Public API ───────────────────────────────────────────────────────────────
def build_styled_gdf(
data_path: str | Path,
rules_path: str | Path,
crs: str = "EPSG:4326",
) -> tuple[gpd.GeoDataFrame, list[dict]]:
"""
Load spatial data, normalize attributes, apply rules, return styled GDF
and legend entries for downstream export.
Args:
data_path: Path to GeoPackage, GeoJSON, or Shapefile.
rules_path: Path to YAML rule file.
crs: Target CRS; input is reprojected if it differs.
Returns:
(styled_gdf, legend_entries)
"""
gdf = gpd.read_file(data_path)
if gdf.crs is None:
raise ValueError("Input data has no CRS — set it explicitly before calling build_styled_gdf.")
if str(gdf.crs) != crs:
gdf = gdf.to_crs(crs)
gdf = normalize_attributes(gdf)
rules = load_rules(Path(rules_path))
styled_gdf = apply_rules(gdf, rules)
legend_entries = extract_legend_entries(rules)
return styled_gdf, legend_entriesPerformance Optimization Patterns
Vectorize attribute normalization. The normalize_attributes function above uses .str.lower() on pandas Series, which is already vectorized. Avoid row-wise apply(lambda r: r["col"].lower()) — it falls back to Python-level iteration and is 10-50× slower on large GeoDataFrames.
Pre-compile constant conditions. For rule sets that never change between features, pre-evaluate any condition that references only literal values at load time. If "True" is your fallback condition, evaluate it once and cache the result rather than passing it through simpleeval for every feature.
Batch rule evaluation with pandas eval. For simple arithmetic comparisons (population > 50000, area_km2 < 10), gdf.eval() uses numexpr under the hood and runs at near-C speed. Reserve simpleeval for compound or string conditions; use gdf.eval() for numeric thresholds:
# Fast path for numeric thresholds
motorway_mask = gdf.eval("lane_count >= 4 and speed_limit > 80")
gdf.loc[motorway_mask, "road_class"] = "motorway"Spatial indexing before attribute joins. When joining external styling metadata to spatial features, use a spatial index (gdf.sindex) to restrict candidates before the attribute merge. STRtree-backed index lookups are O(log N) against O(N) for a brute-force scan. The label collision avoidance algorithms used in dense urban renderings rely on the same STRtree principle for quadtree partitioning.
Common Pitfalls and Debugging
Silent case mismatch in categorical conditions. A rule condition "land_use == 'residential'" fires only when the attribute value is exactly 'residential'. GeoPackages exported from QGIS often preserve original mixed casing. Without normalize_attributes, 'Residential' and 'RESIDENTIAL' fall through to the global fallback, which renders as plain grey — indistinguishable from genuinely unclassified features. Fix: always call normalize_attributes before apply_rules, and add an assertion that checks for unexpected unique values:
expected = {"residential", "commercial", "industrial", "__null__"}
actual = set(gdf["land_use"].unique())
unexpected = actual - expected
if unexpected:
warnings.warn(f"Unexpected land_use values: {unexpected}")Non-deterministic output from duplicate priority integers. Python’s sorted() is stable — it preserves insertion order for equal keys — but insertion order in a YAML mapping is not guaranteed across all parsers. Two rules at priority: 50 may swap positions depending on how PyYAML loads the file. The load_rules function above raises ValueError on duplicate priorities; wire this check into your pre-commit hook so it is impossible to merge a rule file with colliding priorities.
FeatureNotAvailable from typos in condition attribute names. simpleeval raises FeatureNotAvailable when a condition references a name not in the names dict. The current evaluate_condition implementation catches this and warns; in development mode, promote it to an exception to catch typos early. Add a rule-validation step that introspects each condition’s AST for Name nodes and checks them against the GeoDataFrame’s column list.
Memory exhaustion on national-scale datasets. Iterating a 10 M-feature GeoDataFrame row by row in apply_rules will exhaust RAM and take minutes. For datasets above ~500 k features, chunk the GeoDataFrame and process in batches:
CHUNK_SIZE = 100_000
def apply_rules_chunked(
gdf: gpd.GeoDataFrame, rules: list[dict]
) -> gpd.GeoDataFrame:
chunks = [
apply_rules(gdf.iloc[i : i + CHUNK_SIZE], rules)
for i in range(0, len(gdf), CHUNK_SIZE)
]
return gpd.GeoDataFrame(
pd.concat(chunks, ignore_index=True), geometry="geometry", crs=gdf.crs
)Legend entries drifting from rendered output. When rules are edited mid-sprint, the legend extracted from extract_legend_entries reflects the current rule set while cached rendered tiles reflect the previous one. The fix is to invalidate the tile cache whenever the rule YAML changes — use a hash of the rule file as a cache key, and invalidate downstream tiles when it changes. This is the same versioning discipline that dynamic legend generation workflows use to keep legend SVGs in sync with rendered layers.
Validation, Testing, and CI/CD Integration
Treat styling rules as application code. Every rule file should have a corresponding pytest module that exercises its conditions against synthetic GeoDataFrame fixtures:
# tests/test_land_use_rules.py
import geopandas as gpd
import pytest
from shapely.geometry import Point
from rule_engine import build_styled_gdf
@pytest.fixture
def synthetic_gdf():
return gpd.GeoDataFrame(
{
"land_use": ["residential", "COMMERCIAL", "unknown"],
"population": [1200, 8500, 0],
"geometry": [Point(0, 0), Point(1, 0), Point(2, 0)],
},
crs="EPSG:4326",
)
def test_residential_fill(synthetic_gdf, tmp_path):
rules_path = tmp_path / "rules.yaml"
# write minimal rules file ...
styled, _ = build_styled_gdf(synthetic_gdf, rules_path)
assert styled.loc[0, "resolved_style"]["fill"] == "#F0EAD6"
def test_unknown_falls_back_to_default(synthetic_gdf, tmp_path):
styled, _ = build_styled_gdf(synthetic_gdf, tmp_path / "rules.yaml")
# 'unknown' should match the fallback rule
assert styled.loc[2, "resolved_style"]["fill"] == "#E8E8E8"In CI, add a headless rendering step after unit tests: render a fixed-extent test fixture at 1:50 000 and 1:250 000, compare the output PNG against a snapshot baseline using pytest-snapshot. Any visual regression — a colour shift, a missing road class, a changed stroke weight — fails the build before it reaches production.
GitHub Actions configuration (abbreviated):
# .github/workflows/style-engine-ci.yml
jobs:
test-style-engine:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with: { python-version: "3.11" }
- run: pip install -r requirements.txt
- run: python -m jsonschema -i rules/land_use.yaml schema/rule_schema.json
- run: pytest tests/ --snapshot-update=falseConclusion
A rule-based styling engine moves cartographic design from a GUI into a testable, version-controlled Python module. The patterns here — attribute normalization before evaluation, simpleeval sandboxing, priority-unique rule sorting, chunked processing for large datasets, and snapshot-tested CI — give you a pipeline that produces identical output on every run and degrades gracefully when data deviates from expectations. From here, the natural next steps are building a JSON-based rule styling engine for QGIS to validate rules in a desktop environment before deploying them headlessly, and integrating theme inheritance systems so your rule output adapts automatically to light and dark rendering contexts without duplicating rule files.
Related
- Building a JSON-Based Rule Styling Engine for QGIS — Validate and prototype rule schemas inside QGIS before deploying them in headless Python pipelines.
- Label Collision Avoidance Algorithms — STRtree and force-directed approaches for resolving text overlap when rule-assigned symbol sizes shift label anchors.
- Dynamic Legend Generation — Auto-generate legend SVGs from the same rule set that drives rendering, keeping legend and map in guaranteed sync.
- Theme Inheritance Systems — Propagate light/dark theme tokens through rule output without duplicating style logic.