Configuring Valhalla trace_attributes for Fleet Routing

Once Valhalla Meili map matching is returning a clean matched path, the value for fleet analytics lives in the per-edge attributes that /trace_attributes hands back: the posted speed limit on each segment a vehicle drove, the road class (was the truck on a residential street it should not have been?), the surface (paved versus unpaved, which drives wear models), and the turn geometry between edges. This page is the focused, copy-paste implementation for extracting those attributes reliably — a single self-contained class that builds a tuned request, posts it, and flattens the response into a per-edge table. It extends the parent workflow rather than repeating it; if you have not yet built tiles or made a first successful match, start there.

Compatibility and Configuration Requirements

Self-Contained trace_attributes Client

The class below builds a tuned request, posts it, and flattens the edges array into a list of per-edge dictionaries. Every parameter choice is commented inline.

import math
import requests
import pandas as pd
from typing import Optional


class TraceAttributeExtractor:
    """
    Post a tuned Valhalla /trace_attributes request and flatten the matched
    edges into a per-edge analytics table.

    Parameters
    ----------
    base_url : str
        Root URL of the Valhalla service, e.g. 'http://localhost:8002'.
    costing : str
        Costing model: 'auto', 'truck', 'bus', 'bicycle', 'pedestrian'.
        Must match the vehicle class or restrictions are meaningless.
    gps_accuracy : float
        Request-level receiver sigma in metres. Scales the HMM emission term.
        5-8 for clean automotive GPS; 12-20 in dense urban canyons.
        A per-point 'accuracy' on a shape entry overrides this value.
    search_radius : float
        Candidate-edge search cutoff in metres around each point. Must be
        wide enough to contain the true edge given real error: keep it at
        roughly 4-5x gps_accuracy. Too small -> no candidates -> empty match.
    interpolation_distance : float
        Metres below which nearby points are collapsed / interpolated.
        Raise it (to ~30) for sparse traces; lower it for dense 1 Hz traces.
    turn_penalty_factor : float
        Multiplier on turn cost inside the costing model. Higher values make
        the matcher prefer straighter, higher-class roads and suppress
        spurious hops onto side streets. 0 = default costing; 50-500 tightens.
    """

    def __init__(
        self,
        base_url: str,
        costing: str = "auto",
        gps_accuracy: float = 8.0,
        search_radius: float = 40.0,
        interpolation_distance: float = 10.0,
        turn_penalty_factor: float = 0.0,
        truck_options: Optional[dict] = None,
        timeout: float = 30.0,
    ):
        self.base_url = base_url.rstrip("/")
        self.costing = costing
        self.gps_accuracy = gps_accuracy
        self.search_radius = search_radius
        self.interpolation_distance = interpolation_distance
        self.turn_penalty_factor = turn_penalty_factor
        self.truck_options = truck_options or {}
        self.timeout = timeout

    def _costing_options(self) -> dict:
        """Assemble the costing_options block for the selected model."""
        opts = {"turn_penalty_factor": self.turn_penalty_factor}
        if self.costing == "truck":
            # Sensible articulated-truck defaults; override via truck_options
            opts.update({
                "height": 4.11,
                "width": 2.6,
                "length": 21.5,
                "weight": 40.0,
                "axle_load": 9.0,
                "hazmat": False,
            })
            opts.update(self.truck_options)
        return {self.costing: opts}

    def _build_shape(self, df: pd.DataFrame) -> list[dict]:
        """Convert a cleaned, time-sorted ping DataFrame to a shape array."""
        shape = []
        for row in df.itertuples(index=False):
            point = {"lon": float(row.lon), "lat": float(row.lat)}
            if hasattr(row, "epoch_s"):
                point["time"] = int(row.epoch_s)          # sharpens transitions
            acc = getattr(row, "horiz_acc_m", None)
            if acc is not None and acc == acc:             # not NaN
                point["accuracy"] = max(float(acc), 4.0)   # floor at 4 m
            shape.append(point)
        return shape

    def match(self, df: pd.DataFrame) -> dict:
        """POST the trace and return the raw Valhalla JSON response."""
        payload = {
            "shape": self._build_shape(df),
            "costing": self.costing,
            "shape_match": "map_snap",
            "costing_options": self._costing_options(),
            "trace_options": {
                "gps_accuracy": self.gps_accuracy,
                "search_radius": self.search_radius,
                "interpolation_distance": self.interpolation_distance,
            },
            "filters": {
                "attributes": [
                    "edge.way_id", "edge.names", "edge.road_class",
                    "edge.speed_limit", "edge.speed", "edge.surface",
                    "edge.length", "edge.begin_heading", "edge.end_heading",
                    "matched.point", "matched.type", "matched.edge_index",
                    "confidence_score",
                ],
                "action": "include",     # return ONLY the listed attributes
            },
        }
        resp = requests.post(
            f"{self.base_url}/trace_attributes", json=payload, timeout=self.timeout
        )
        resp.raise_for_status()
        return resp.json()

    @staticmethod
    def _turn_degree(prev_end: Optional[float], cur_begin: Optional[float]) -> Optional[float]:
        """Signed turn angle in degrees between two edges at their shared node."""
        if prev_end is None or cur_begin is None:
            return None
        delta = (cur_begin - prev_end + 180.0) % 360.0 - 180.0   # normalise to [-180, 180)
        return round(delta, 1)

    def to_edge_table(self, result: dict) -> pd.DataFrame:
        """Flatten the matched edges into one row per edge with turn geometry."""
        edges = result.get("edges", [])
        rows = []
        prev_end_heading = None
        for i, edge in enumerate(edges):
            begin_h = edge.get("begin_heading")
            rows.append({
                "edge_index": i,
                "way_id": edge.get("way_id"),
                "name": (edge.get("names") or [None])[0],
                "road_class": edge.get("road_class"),
                # Posted limit where OSM tags it; fall back to estimated speed
                "speed_limit_kmh": edge.get("speed_limit") or edge.get("speed"),
                "speed_source": "posted" if edge.get("speed_limit") else "estimated",
                "surface": edge.get("surface"),
                "length_km": edge.get("length"),
                "turn_degree": self._turn_degree(prev_end_heading, begin_h),
            })
            prev_end_heading = edge.get("end_heading")
        df = pd.DataFrame(rows)
        df.attrs["confidence_score"] = result.get("confidence_score")
        return df


# ---------------------------------------------------------------------------
# Usage
# ---------------------------------------------------------------------------
# extractor = TraceAttributeExtractor(
#     base_url="http://localhost:8002",
#     costing="truck",
#     gps_accuracy=10.0,
#     search_radius=50.0,
#     turn_penalty_factor=100.0,
#     truck_options={"height": 4.0, "weight": 26.0},
# )
# result = extractor.match(cleaned_pings_df)   # df cols: lon, lat, epoch_s, horiz_acc_m
# edge_table = extractor.to_edge_table(result)
# print(edge_table.head(), edge_table.attrs["confidence_score"])

Execution and Tuning Guidelines

Run match() on a single trip’s worth of cleaned, time-sorted pings, then to_edge_table() to get the analytics table. The four tuning knobs interact, so change one at a time and inspect the confidence_score and the edge table together:

• gps_accuracy — the emission sigma. Raise it (toward 15–20 m) when good matches are being penalised with low confidence in urban canyons; lower it (toward 5 m) with clean open-sky data to make the matcher discriminate more sharply between nearby parallel roads. Setting it far below the true error makes every emission improbable and tanks confidence_score even on a correct path.
• search_radius — the candidate cutoff. Raise it when the response is empty or points snap to the wrong side of a divided highway; lower it if matching is slow because too many candidates are considered per point. Keep it at roughly 4–5 × gps_accuracy so the true edge cannot fall outside the window.
• interpolation_distance — raise it (to ~30 m) for sparse traces so intermediate shape points are synthesised along candidate routes, filling gaps; lower it for dense 1 Hz data where over-interpolation just adds noise.
• turn_penalty_factor — raise it (100–500) to suppress spurious hops onto side streets and keep the match on straighter, higher-class roads; lower it toward 0 if the matcher is now cutting across genuine sharp turns at junctions. This flows into the HMM transition term, so it changes which edges win, not just the reported cost.

The speed_source column matters for speed profiling from raw GPS coordinates: a posted limit is authoritative for compliance checks, whereas an estimated value is Valhalla’s modelled free-flow speed and should not be reported as a legal limit. The turn_degree column feeds harsh-cornering and route-shape analytics without a second geometry pass.

Common Pitfalls

Up: Valhalla Meili Map Matching for Telematics | Routing Engine Integration for Fleet Telematics

Related

• Valhalla Meili Map Matching for Telematics — the parent workflow: building tiles, the HMM model, and the full trace_attributes call
• Hidden Markov Model Map Matching in Python — how the emission and transition probabilities that costing and accuracy tune actually work
• Speed Profiling from Raw GPS Coordinates — consuming the extracted speed_limit and speed_source columns for compliance and behaviour scoring
• Choosing a Routing Engine: OSRM vs Valhalla vs GraphHopper — where Valhalla’s per-edge attribute richness fits against the alternatives
• Valhalla documentation — the official trace_attributes attribute list and costing reference