Interpolating GPS Gaps During Tunnel Signal Loss

A delivery route through a mile-long tunnel, an urban canyon between tall buildings, or an underground parking structure produces the same signature in raw telematics data: a run of fixes at the normal reporting interval, then a gap of several seconds to a few minutes, then fixes resuming as if nothing happened. This extends Timestamp Synchronization for Multi-Device GPS Logs with the specific problem of what to do with that gap — leave it as a hole in the trace, or fill it with a plausible synthetic path without inventing movement the vehicle never made.

The naive fix — linear interpolation between the last fix before the gap and the first fix after it — is fine for a short, straight gap and actively wrong for a curving tunnel or a gap that contains a genuine stop. The right approach depends on gap duration, road geometry, and how much heading/speed context is available on either side, which is why this page treats interpolation strategy as a tunable choice rather than a fixed rule.

Compatibility and Configuration

Tables scroll horizontally on narrow viewports.

A Self-Contained Gap Interpolator

The class below detects gaps beyond a configurable threshold, checks that filling them would not fabricate an implausible distance, and fills the gap with one of three strategies — always marking inserted rows so they can be excluded downstream.

from __future__ import annotations

from dataclasses import dataclass
from typing import Literal

import numpy as np
import pandas as pd
from pyproj import Geod
from scipy.interpolate import CubicSpline

FillMethod = Literal["linear", "spline", "dead_reckoning"]

_GEOD = Geod(ellps="WGS84")


@dataclass
class TunnelGapInterpolator:
    """
    Detect multi-second GPS signal-loss gaps (tunnels, urban canyons, parking
    structures) and fill them with a chosen interpolation strategy, flagging
    every fabricated point so downstream consumers can distinguish measured
    from synthetic positions.

    Parameters
    ----------
    max_gap_seconds : float
        Any consecutive-fix interval strictly greater than this is treated
        as a signal-loss gap and becomes a fill candidate. 10-12 s is a
        reasonable default for a 5 s fleet reporting interval; raise it if
        your normal sampling already has occasional multi-second jitter you
        do not want flagged as tunnels.
    target_interval_s : float
        Spacing between synthetic points inserted into a filled gap. Match
        it to the fleet's normal reporting interval so downstream speed and
        heading derivations do not see an artificial frequency change at the
        gap boundary.
    method : FillMethod
        'linear' draws a straight line between the fixes bracketing the
        gap. 'spline' fits a cubic spline through `context_points` real
        fixes on each side, so the fill respects the trajectory's curvature
        and heading. 'dead_reckoning' projects forward from the last known
        heading/speed and backward from the next known heading/speed,
        blending the two paths across the gap.
    max_interp_distance_m : float
        If the great-circle distance between the fixes bracketing a gap
        exceeds this value, the gap is left unfilled rather than fabricating
        a long synthetic path. Protects against silently drawing a straight
        line across a gap that actually contains a real stop or a
        multi-kilometre detour.
    flag_column : str
        Name of the boolean column added to mark synthetic rows. Carry this
        column through every downstream schema so stop detection and other
        position-sensitive analytics can exclude fabricated points.
    context_points : int
        Number of real fixes on each side of a gap used to fit the spline or
        estimate heading/speed for dead reckoning. 3-5 gives a stable
        estimate without over-fitting to a single noisy fix.
    """

    max_gap_seconds: float = 10.0
    target_interval_s: float = 5.0
    method: FillMethod = "spline"
    max_interp_distance_m: float = 3000.0
    flag_column: str = "is_synthetic"
    context_points: int = 4

    def fill(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Detect and fill signal-loss gaps in a single device's GPS stream.

        Parameters
        ----------
        df : DataFrame with 'timestamp' (tz-aware), 'latitude', 'longitude'
             for exactly one device, in any order.

        Returns
        -------
        DataFrame sorted by timestamp with synthetic rows inserted where
        gaps were filled, a boolean `flag_column`, and a `gap_seconds`
        column recording the original gap length for every synthetic row
        (NaN for real fixes, and for gaps skipped by max_interp_distance).
        """
        df = df.sort_values("timestamp").reset_index(drop=True)
        df[self.flag_column] = False
        df["gap_seconds"] = np.nan

        gap_seconds = df["timestamp"].diff().dt.total_seconds()
        gap_starts = df.index[gap_seconds > self.max_gap_seconds].tolist()

        pieces = []
        cursor = 0
        for idx in gap_starts:
            pieces.append(df.iloc[cursor:idx])
            before = df.iloc[idx - 1]
            after = df.iloc[idx]
            span_s = float(gap_seconds.iloc[idx])
            if self._distance_m(before, after) > self.max_interp_distance_m:
                # Do not fabricate a path across an implausibly large gap.
                cursor = idx
                continue
            pieces.append(self._fill_gap(df, idx, span_s))
            cursor = idx
        pieces.append(df.iloc[cursor:])

        return pd.concat(pieces, ignore_index=True)

    @staticmethod
    def _distance_m(before: pd.Series, after: pd.Series) -> float:
        _, _, dist = _GEOD.inv(
            before["longitude"], before["latitude"],
            after["longitude"], after["latitude"],
        )
        return float(dist)

    def _fill_gap(self, df: pd.DataFrame, idx: int, span_s: float) -> pd.DataFrame:
        before = df.iloc[idx - 1]
        after = df.iloc[idx]
        n_points = max(int(span_s // self.target_interval_s) - 1, 1)
        offsets = np.linspace(0, span_s, n_points + 2)[1:-1]
        timestamps = [before["timestamp"] + pd.Timedelta(seconds=o) for o in offsets]

        if self.method == "linear":
            lats, lons = self._linear_fill(before, after, offsets, span_s)
        elif self.method == "spline":
            lats, lons = self._spline_fill(df, idx, offsets)
        else:
            lats, lons = self._dead_reckoning_fill(df, idx, offsets, span_s)

        synthetic = pd.DataFrame({
            "timestamp": timestamps,
            "latitude": lats,
            "longitude": lons,
        })
        synthetic[self.flag_column] = True
        synthetic["gap_seconds"] = span_s
        return synthetic

    @staticmethod
    def _linear_fill(before, after, offsets, span_s):
        frac = offsets / span_s
        lats = before["latitude"] + frac * (after["latitude"] - before["latitude"])
        lons = before["longitude"] + frac * (after["longitude"] - before["longitude"])
        return lats, lons

    def _spline_fill(self, df: pd.DataFrame, idx: int, offsets: np.ndarray):
        n = self.context_points
        ctx = pd.concat([
            df.iloc[max(0, idx - n):idx],
            df.iloc[idx:idx + n],
        ])
        t0 = df.iloc[idx - 1]["timestamp"]
        t = (ctx["timestamp"] - t0).dt.total_seconds().to_numpy()
        order = np.argsort(t)
        t_sorted = t[order]
        lat_sorted = ctx["latitude"].to_numpy()[order]
        lon_sorted = ctx["longitude"].to_numpy()[order]
        lat_spline = CubicSpline(t_sorted, lat_sorted)
        lon_spline = CubicSpline(t_sorted, lon_sorted)
        return lat_spline(offsets), lon_spline(offsets)

    def _dead_reckoning_fill(self, df: pd.DataFrame, idx: int, offsets: np.ndarray, span_s: float):
        n = self.context_points
        before = df.iloc[idx - 1]
        after = df.iloc[idx]
        prior_ctx = df.iloc[max(0, idx - n):idx]
        next_ctx = df.iloc[idx:idx + n].iloc[::-1]

        fwd_bearing, fwd_speed = self._heading_speed(prior_ctx)
        back_bearing, back_speed = self._heading_speed(next_ctx)

        fwd_lats, fwd_lons, back_lats, back_lons = [], [], [], []
        for offset in offsets:
            flon, flat, _ = _GEOD.fwd(
                before["longitude"], before["latitude"], fwd_bearing, fwd_speed * offset
            )
            fwd_lats.append(flat)
            fwd_lons.append(flon)

            remaining = span_s - offset
            blon, blat, _ = _GEOD.fwd(
                after["longitude"], after["latitude"],
                (back_bearing + 180.0) % 360.0, back_speed * remaining,
            )
            back_lats.append(blat)
            back_lons.append(blon)

        frac = offsets / span_s
        lats = (1 - frac) * np.array(fwd_lats) + frac * np.array(back_lats)
        lons = (1 - frac) * np.array(fwd_lons) + frac * np.array(back_lons)
        return lats, lons

    @staticmethod
    def _heading_speed(ctx: pd.DataFrame) -> tuple[float, float]:
        """Average bearing (degrees) and speed (m/s) across a short run of fixes."""
        if len(ctx) < 2:
            return 0.0, 0.0
        lats = ctx["latitude"].to_numpy()
        lons = ctx["longitude"].to_numpy()
        ts = ctx["timestamp"].to_numpy()
        bearings, speeds = [], []
        for i in range(len(ctx) - 1):
            az, _, dist = _GEOD.inv(lons[i], lats[i], lons[i + 1], lats[i + 1])
            dt = (ts[i + 1] - ts[i]) / np.timedelta64(1, "s")
            if dt > 0:
                bearings.append(az % 360.0)
                speeds.append(dist / dt)
        if not bearings:
            return 0.0, 0.0
        return float(np.mean(bearings)), float(np.mean(speeds))

Usage against a device stream with a 40-second tunnel gap:

interpolator = TunnelGapInterpolator(
    max_gap_seconds=10.0,
    method="dead_reckoning",
    max_interp_distance_m=2000.0,
)
filled = interpolator.fill(device_df)

# Exclude fabricated points before any stop-detection pass.
measured_only = filled[~filled["is_synthetic"]]

Execution and Tuning Guidelines

max_gap_seconds. Set this above your fleet’s normal reporting interval, not at it. A fleet reporting every 5 seconds over an intermittently congested cellular network will produce occasional 6-8 second gaps that are packet loss, not tunnels. A threshold of 10-12 seconds filters those out while still catching genuine short tunnels. Fleets on a coarser 30-second interval should scale the threshold up proportionally — comparing a 30 s reporting fleet against a fixed 10 s threshold will flag almost every normal interval as a gap.

method. linear is the cheapest and is defensible only for short gaps (under roughly 15-20 seconds) where the road is close to straight — a short underpass, a brief line-of-sight loss between buildings. spline is the safer general default: it uses the heading implied by the last few real fixes on each side, so it curves through a bend a straight line would cut across. dead_reckoning is the most physically grounded when both the entry and exit heading are stable and roughly aligned, which is common in long highway tunnels but not in a multi-turn underground parking garage.

max_interp_distance_m. This is the guard against fabricating a long synthetic path. A gap that implies 3+ km of travel is either an extended dropout (multi-mile tunnel, ferry crossing) or — more often in production data — a device that went offline for an extended period, possibly including a genuine stop. Set this in proportion to plausible tunnel/dropout length for your operating region; interstate tunnels rarely exceed 2-3 km, so a 3,000 m default catches most real tunnels while refusing to fill a gap caused by a device going dark for an hour.

flag_column. Never drop this column downstream. Speed profiling can tolerate synthetic points because a continuous trace is more useful than a hole, but stop detection should filter them out explicitly — a synthetic point can never represent a genuine dwell event, and DBSCAN-style clustering has no way to know the difference unless the flag is there to filter on.

Once gaps are filled, the stream is ready for the same Kalman filtering for GPS noise reduction pass used on the rest of the trip — but consider widening the filter’s process noise around flagged synthetic segments, since the filter’s confidence in a fabricated position should be lower than its confidence in a measured one.

Common Pitfalls

Up: Timestamp Synchronization for Multi-Device GPS Logs | GPS Data Preprocessing & Cleaning Fundamentals

Related

• Timestamp Synchronization for Multi-Device GPS Logs — the parent workflow for aligning device clocks before gap detection runs
• How to Align GPS Timestamps Across Mixed OBD-II and Mobile Devices — device-specific clock calibration that should run before gap detection to avoid false gaps from drift
• Kalman Filtering for GPS Noise Reduction — smooth the combined measured-plus-synthetic stream, widening process noise across flagged segments
• Stop Detection & Dwell Time Analytics — exclude synthetic points before clustering so fabricated positions cannot register as a stop
• GPS Data Preprocessing & Cleaning Fundamentals — the full preprocessing pipeline this gap-fill step belongs to