Batch Map Matching with GraphHopper in Python

Matching one trace against a GraphHopper /match server is a single HTTP call, covered in GraphHopper map matching for fleet routing. The problem this page solves is the next one: reprocessing an entire day of fleet telemetry — tens of thousands of per-trip traces — through that same endpoint without melting the server, losing partial progress, or turning a handful of broken sequences into a failed batch. The answer is a bounded thread pool with retry that separates transient failures from deterministic ones, plus aggregation into a pandas DataFrame so the output is a report, not a pile of JSON. The class below is self-contained and copy-paste ready; every knob is explained inline and in the tuning section that follows.

Compatibility and Configuration Requirements

Tables scroll horizontally on narrow viewports. See the GraphHopper documentation for the authoritative endpoint contract and the GraphHopper repository for the map-matching module.

A Self-Contained Batch Matcher

import time
import logging
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from datetime import timezone
from concurrent.futures import ThreadPoolExecutor, as_completed


import requests
import pandas as pd
from requests.adapters import HTTPAdapter

logger = logging.getLogger("gh_batch")


@dataclass
class Trace:
    """One ordered vehicle trace destined for a single /match call."""
    trace_id: str                 # unique key, e.g. f"{vehicle_id}:{trip_id}"
    points: list                  # [{"lat": float, "lon": float, "time": datetime}, ...]
    profile: str = "car"          # must be declared in the server config.yml


class GraphHopperBatchMatcher:
    """
    Batch map matcher for a self-hosted GraphHopper /match server.

    Parameters
    ----------
    base_url : str
        Root of the GraphHopper server, e.g. "http://localhost:8989".
    gps_accuracy : float
        Emission std-dev in metres. Set to the receiver's true ~1-sigma error:
        8-12 for clean automotive GPS, 25-40 for sparse or urban-canyon traces.
        Too small breaks sequences; too large lets the match wander.
    max_visited_nodes : int
        Transition search budget. Raise (5000-10000) for sparse traces; capping
        it protects p99 latency from a single pathological trace.
    workers : int
        Thread-pool size. Scale to the SERVER core count (2-4 per core), not the
        client's -- matching is CPU-bound on one shared graph.
    chunk_size : int
        Traces submitted per wave. Bounds client memory and gives checkpointable
        progress on large overnight batches.
    max_retries : int
        Retry attempts for TRANSIENT failures only (timeout, 5xx, conn reset).
        Deterministic 400 broken-sequence errors are never retried.
    backoff_base : float
        Seconds for exponential backoff: sleep = backoff_base * 2 ** attempt.
    request_timeout : float
        Per-request timeout in seconds. Must exceed the slowest legitimate trace.
    """

    def __init__(
        self,
        base_url: str,
        gps_accuracy: float = 12.0,
        max_visited_nodes: int = 5_000,
        workers: int = 8,
        chunk_size: int = 500,
        max_retries: int = 3,
        backoff_base: float = 0.5,
        request_timeout: float = 60.0,
    ):
        self.base_url = base_url.rstrip("/")
        self.gps_accuracy = gps_accuracy
        self.max_visited_nodes = max_visited_nodes
        self.workers = workers
        self.chunk_size = chunk_size
        self.max_retries = max_retries
        self.backoff_base = backoff_base
        self.request_timeout = request_timeout

        # One pooled Session shared across threads; sized to the worker count so
        # connections are reused instead of re-opened per request.
        self.session = requests.Session()
        adapter = HTTPAdapter(pool_connections=workers, pool_maxsize=workers)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    # -- GPX serialisation -------------------------------------------------
    @staticmethod
    def _to_gpx(points: list) -> bytes:
        gpx = ET.Element("gpx", version="1.1", creator="gh-batch")
        seg = ET.SubElement(ET.SubElement(gpx, "trk"), "trkseg")
        for p in points:
            pt = ET.SubElement(
                seg, "trkpt", lat=f"{p['lat']:.6f}", lon=f"{p['lon']:.6f}"
            )
            t = p["time"].astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
            ET.SubElement(pt, "time").text = t
        return ET.tostring(gpx, encoding="utf-8", xml_declaration=True)

    # -- single request with retry ----------------------------------------
    def _match_one(self, trace: Trace) -> dict:
        """Match one trace. Retries transient errors; surfaces 400s as failures."""
        params = {
            "profile": trace.profile,
            "gps_accuracy": self.gps_accuracy,
            "max_visited_nodes": self.max_visited_nodes,
            "points_encoded": "false",
            "details": ["road_class"],
        }
        body = self._to_gpx(trace.points)
        headers = {"Content-Type": "application/gpx+xml"}

        for attempt in range(self.max_retries + 1):
            try:
                resp = self.session.post(
                    f"{self.base_url}/match",
                    params=params,
                    data=body,
                    headers=headers,
                    timeout=self.request_timeout,
                )
            except (requests.Timeout, requests.ConnectionError) as exc:
                if attempt < self.max_retries:
                    time.sleep(self.backoff_base * 2 ** attempt)
                    continue
                return self._fail(trace, f"transport: {exc}")

            if resp.status_code == 200:
                return self._parse(trace, resp.json())

            # 4xx (e.g. broken sequence) is deterministic -- do not retry.
            if 400 <= resp.status_code < 500:
                return self._fail(trace, f"client {resp.status_code}: {resp.text[:120]}")

            # 5xx is transient -- back off and retry.
            if attempt < self.max_retries:
                time.sleep(self.backoff_base * 2 ** attempt)
                continue
            return self._fail(trace, f"server {resp.status_code}")

        return self._fail(trace, "exhausted retries")

    # -- result shaping ----------------------------------------------------
    @staticmethod
    def _parse(trace: Trace, payload: dict) -> dict:
        path = payload["paths"][0]
        return {
            "trace_id": trace.trace_id,
            "profile": trace.profile,
            "status": "ok",
            "matched_distance_m": path["distance"],
            "matched_time_s": path["time"] / 1000.0,
            "point_count": len(trace.points),
            "error": None,
        }

    @staticmethod
    def _fail(trace: Trace, error: str) -> dict:
        return {
            "trace_id": trace.trace_id,
            "profile": trace.profile,
            "status": "failed",
            "matched_distance_m": None,
            "matched_time_s": None,
            "point_count": len(trace.points),
            "error": error,
        }

    # -- public entry point ------------------------------------------------
    def run(self, traces: list) -> pd.DataFrame:
        """Match every trace and return one aggregated row per trace."""
        rows: list = []
        for start in range(0, len(traces), self.chunk_size):
            chunk = traces[start:start + self.chunk_size]
            with ThreadPoolExecutor(max_workers=self.workers) as pool:
                futures = {pool.submit(self._match_one, t): t for t in chunk}
                for fut in as_completed(futures):
                    rows.append(fut.result())
            logger.info("matched %d/%d traces", len(rows), len(traces))
        return pd.DataFrame(rows)


# ---------------------------------------------------------------------------
# Usage
# ---------------------------------------------------------------------------
# matcher = GraphHopperBatchMatcher(
#     base_url="http://localhost:8989",
#     gps_accuracy=12.0,
#     workers=8,           # server has 4 physical cores -> 2 workers/core
#     chunk_size=500,
# )
# traces = [Trace(trace_id=f"{v}:{trip}", points=pts, profile="truck")
#           for (v, trip, pts) in grouped_trips]
# results = matcher.run(traces)
# print(results["status"].value_counts())
# results.to_parquet("match_results.parquet")

Execution and Tuning Guidelines

workers — size to the server, not the client

The most common mistake is setting workers from the client’s core count. GraphHopper matches against one shared in-memory graph and is CPU-bound, so throughput is capped by the server’s physical cores. Two to four workers per server core saturates it; more just queues requests inside the JVM and lengthens the latency tail. Raise workers while watching p99 latency and stop as soon as the tail degrades — that inflection is your ceiling.

gps_accuracy — the emission window

gps_accuracy is the standard deviation of the measurement model in metres, identical to the parameter on the single-trace endpoint. Raise it for sparse or noisy fleets so legitimate edges are not penalised out of the emission window; lower it for clean automotive GPS so the match does not wander onto parallel roads. Because a batch usually mixes receiver types, carry the value per trace when your fleets differ rather than forcing one global constant.

max_visited_nodes — the transition budget

This bounds the shortest-path search that connects consecutive candidates. Sparse traces need a larger budget to bridge the gap between fixes; raising it lifts recall on 30–60 s sampling but lets a single pathological trace dominate latency. Set it high enough to clear your normal sparse traces (5,000–10,000) and cap it there, then handle genuine signal-loss gaps by splitting the trace upstream via interpolating GPS gaps during tunnel signal loss.

chunk_size — memory and checkpointing

Chunking submits traces in waves instead of holding one future per trace in memory. It bounds client RSS on large batches, and — because each chunk completes before the next starts — it gives you a natural checkpoint boundary. Persist results per chunk (append to Parquet) so a crash three hours into an overnight run resumes from the last completed chunk rather than re-matching everything.

Retry policy — separate transient from deterministic

The retry logic deliberately treats a 400 differently from a 5xx. A broken-sequence 400 is deterministic: the identical trace and parameters will fail identically, so retrying only burns server capacity. Route those to a fallback that cleans outliers with Kalman filtering, splits at gaps, and re-submits with a wider gps_accuracy. Reserve exponential backoff for timeouts, connection resets, and 5xx — the failures that a moment later may succeed.

Common Pitfalls

Up: GraphHopper Map Matching for Fleet Routing | Routing Engine Integration for Fleet Telematics

Related

• GraphHopper Map Matching for Fleet Routing — the parent topic covering OSM import, custom truck profiles, and the single-trace /match contract this batcher calls
• Handling Routing Engine Rate Limits with Circuit Breakers — protect a CPU-bound match server from a thread pool that would otherwise overwhelm it
• Hidden Markov Model Map Matching in Python — the emission and transition model that gps_accuracy and max_visited_nodes tune
• Kalman Filtering for GPS Noise Reduction — the fallback cleanup for traces that return a deterministic broken-sequence error