Transport Mode Inference from GPS Speed Signatures

This page extends Step 2 of the multi-modal route matching workflow — mode classification and metadata fusion — with a self-contained implementation of the kinematic side of that fusion: turning a raw speed and acceleration signal into a per-segment mode label without relying on vehicle metadata at all. That matters whenever metadata is missing, stale, or simply wrong (a courier fleet that mixes cargo bikes and vans under one fleet ID, a shared-mobility operator with no declared vehicle class), and it matters as a sanity check against metadata that fusion pipeline trusts too readily. The technique builds directly on the speed profiling pipeline for the underlying velocity signal and produces output compatible with the heading synchronization module’s quality-flag convention.

Compatibility & Configuration Requirements

Production-Ready Implementation

The class below is self-contained: it segments a trace, computes four kinematic features per segment, and classifies mode using rule thresholds. Each threshold is documented inline with the reasoning behind its value.

import numpy as np
import pandas as pd
from dataclasses import dataclass, field


@dataclass
class ModeThresholds:
    """
    Mean-speed and acceleration-variance bands (m/s and m/s^2) used by the
    rule classifier. Values are starting points for road-vehicle fleets
    with occasional cyclists and pedestrians mixed in; recalibrate against
    labelled ground truth before production use.
    """
    walk_max_ms: float = 2.2
    """Upper mean-speed bound for 'walk'. ~7.9 km/h covers brisk walking."""

    bike_max_ms: float = 7.0
    """Upper mean-speed bound for 'bike'. ~25 km/h covers e-bikes."""

    rail_min_ms: float = 12.0
    """Lower mean-speed bound that, combined with near-zero acceleration
    variance and near-zero stop rate, indicates rail rather than road."""

    car_accel_var_max: float = 1.8
    """Acceleration-variance ceiling (m/s^2 squared) for smooth road travel.
    Above this, stop-and-go behaviour dominates the segment."""

    stop_speed_ms: float = 0.5
    """Speed floor below which a ping counts toward stop_rate. Matches the
    jitter gate used in the speed-profiling pipeline."""


class SpeedSignatureModeClassifier:
    """
    Segment a GPS trace and classify travel mode from speed/acceleration
    signatures alone — no vehicle metadata required.

    Parameters
    ----------
    segment_window_s : float
        Length of each non-overlapping segment in seconds. 30 s is a
        reasonable default: long enough to average out GPS jitter, short
        enough to localize a mode change to one or two segments.
    speed_percentile : float
        Percentile used for the "peak speed" feature (default 95th).
        The 95th percentile is more robust to single-ping GPS spikes than
        the maximum while still capturing genuine bursts of speed.
    thresholds : ModeThresholds
        Mode-specific speed and acceleration-variance bands.
    min_segment_pings : int
        Segments with fewer pings than this are labelled 'unknown' rather
        than forced into a class — short segments produce unstable
        percentile and variance estimates.
    """

    def __init__(
        self,
        segment_window_s: float = 30.0,
        speed_percentile: float = 95.0,
        thresholds: ModeThresholds | None = None,
        min_segment_pings: int = 4,
    ):
        self.segment_window_s = segment_window_s
        self.speed_percentile = speed_percentile
        self.thresholds = thresholds or ModeThresholds()
        self.min_segment_pings = min_segment_pings

    def _segment_ids(self, df: pd.DataFrame) -> np.ndarray:
        """Assign a segment id to each row by floor-dividing elapsed seconds."""
        elapsed_s = (df["timestamp"] - df["timestamp"].iloc[0]).dt.total_seconds()
        return np.floor(elapsed_s / self.segment_window_s).astype(int)

    def extract_features(self, segment_df: pd.DataFrame) -> dict:
        """
        Compute the four kinematic features for one segment.

        Returns
        -------
        dict with keys: mean_speed_ms, p95_speed_ms, accel_variance,
        stop_rate, n_pings.
        """
        speed = segment_df["speed_ms"].to_numpy()
        dt_s = (
            segment_df["timestamp"].diff().dt.total_seconds().to_numpy()[1:]
        )
        dt_s = np.where(dt_s <= 0, np.nan, dt_s)
        accel = np.diff(speed) / dt_s

        return {
            "mean_speed_ms": float(np.nanmean(speed)) if len(speed) else 0.0,
            "p95_speed_ms": float(np.nanpercentile(speed, self.speed_percentile))
            if len(speed) else 0.0,
            "accel_variance": float(np.nanvar(accel)) if len(accel) else 0.0,
            "stop_rate": float(
                np.mean(speed < self.thresholds.stop_speed_ms)
            ) if len(speed) else 0.0,
            "n_pings": int(len(segment_df)),
        }

    def classify_segment(self, features: dict) -> str:
        """
        Apply mode-specific rule thresholds to one feature vector.

        Priority order matters: rail is checked first because its speed
        range overlaps with car/truck but its acceleration variance and
        stop rate are near zero. Everything that fails every rule falls
        through to 'car' as the default road-vehicle assumption.
        """
        t = self.thresholds
        mean_v = features["mean_speed_ms"]
        accel_var = features["accel_variance"]
        stop_rate = features["stop_rate"]

        if features["n_pings"] < self.min_segment_pings:
            return "unknown"

        if mean_v >= t.rail_min_ms and accel_var < 0.3 and stop_rate < 0.02:
            return "rail"
        if mean_v <= t.walk_max_ms:
            return "walk"
        if mean_v <= t.bike_max_ms and accel_var < 1.2:
            return "bike"
        if accel_var >= t.car_accel_var_max and stop_rate >= 0.15:
            # Frequent stop-start bursts at road speed: heavier vehicle
            # or stop-and-go traffic rather than free-flowing car travel.
            return "truck"
        return "car"

    def classify_trace(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Full pipeline: segment, extract features, classify.

        Parameters
        ----------
        df : DataFrame with columns 'timestamp' (UTC, sorted) and
             'speed_ms' — the smoothed speed output of the speed-profiling
             pipeline for a single device.

        Returns
        -------
        DataFrame with one row per segment: segment_id, start_time,
        end_time, the four features, and the assigned mode.
        """
        df = df.sort_values("timestamp").reset_index(drop=True)
        df["_segment_id"] = self._segment_ids(df)

        rows = []
        for seg_id, seg_df in df.groupby("_segment_id"):
            features = self.extract_features(seg_df)
            mode = self.classify_segment(features)
            rows.append(
                {
                    "segment_id": int(seg_id),
                    "start_time": seg_df["timestamp"].iloc[0],
                    "end_time": seg_df["timestamp"].iloc[-1],
                    **features,
                    "mode": mode,
                }
            )
        return pd.DataFrame(rows)

Note: training a light classifier instead of hard thresholds

Once a few hundred labelled segments exist — from driver confirmation, GTFS overlap for rail, or manual QA — replace classify_segment with a trained model using the same four features. A shallow classifier generalizes across fleets better than hand-tuned bands, particularly at the bike/car boundary where e-bike and low-power moped speeds overlap:

from sklearn.ensemble import RandomForestClassifier

FEATURE_COLUMNS = ["mean_speed_ms", "p95_speed_ms", "accel_variance", "stop_rate"]

def train_mode_classifier(
    labelled_segments: pd.DataFrame,
) -> RandomForestClassifier:
    """
    labelled_segments must contain FEATURE_COLUMNS plus a 'mode' ground-
    truth column. max_depth=6 and n_estimators=200 keep inference latency
    low enough for batch scoring of millions of segments.
    """
    clf = RandomForestClassifier(
        n_estimators=200,
        max_depth=6,
        class_weight="balanced",
        random_state=42,
    )
    clf.fit(labelled_segments[FEATURE_COLUMNS], labelled_segments["mode"])
    return clf

Execution & Tuning Guidelines

Run the classifier per device: SpeedSignatureModeClassifier().classify_trace(df) returns one row per segment with the assigned mode. Feed the resulting mode column directly into Step 3 (dynamic graph filtering) of the multi-modal matching pipeline.

Calibrate ModeThresholds per fleet geography before deployment. A fleet operating in a flat, low-traffic suburb will see cleaner separation between bike and car mean speeds than one operating in a dense city centre where congested car segments regularly dip into the bike speed band — this is the central limitation the pitfalls below address.

Common Pitfalls

Up: Multi-Modal Route Matching for Mixed Fleets | Trajectory Analysis & Map Matching Techniques

Related

• Multi-Modal Route Matching for Mixed Fleets — the parent workflow this mode classifier plugs into ahead of graph filtering
• Speed Profiling from Raw GPS Coordinates — supplies the smoothed speed_ms column this classifier depends on
• Calculating Instantaneous vs Average Speed from GPS Traces — background on why instantaneous speed is the correct input for kinematic feature extraction
• Directionality & Heading Synchronization — heading stability is a secondary signal worth fusing in for rail vs road disambiguation
• Hidden Markov Model Map Matching in Python — the mode-filtered graph produced downstream feeds directly into HMM candidate generation