Detecting Harsh Braking Events from Speed Profiles

This page extends the speed profiling cluster with the differentiation step most fleet safety programs need next: turning a clean speed_kmh_smoothed series into discrete, debounced harsh-braking and harsh-acceleration events for driver scoring. The distinction from ordinary velocity analysis matters because acceleration is the second derivative of position — every noise-amplification problem that affects instantaneous speed is squared again when computing its derivative, which makes smoothing order and sampling-rate awareness far less optional here than upstream in the pipeline.

Compatibility & Configuration Requirements

Production-Ready Implementation

The class below computes acceleration from an already-smoothed speed series, flags samples past a deceleration threshold, and debounces consecutive flags into discrete events with a severity grade.

import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import List


@dataclass
class HarshEvent:
    event_type: str          # 'harsh_braking' or 'harsh_acceleration'
    start_time: pd.Timestamp
    end_time: pd.Timestamp
    peak_accel_ms2: float    # most negative (braking) or most positive value
    severity: str            # 'moderate' or 'severe'
    n_samples: int


class HarshEventDetector:
    """
    Detect discrete harsh braking / harsh acceleration events from a
    smoothed GPS speed profile.

    Parameters
    ----------
    decel_threshold_ms2 : float
        Deceleration magnitude (negative, m/s^2) at or below which a
        sample is a harsh-braking candidate. -3.0 m/s^2 (~0.3g) is a
        common fleet-safety starting point.
    accel_threshold_ms2 : float
        Acceleration magnitude (positive, m/s^2) at or above which a
        sample is a harsh-acceleration candidate. 2.5 m/s^2 is a
        reasonable default for loaded commercial vehicles.
    severe_decel_ms2 : float
        Deceleration magnitude beyond which an event is graded 'severe'
        rather than 'moderate'. -5.0 m/s^2 (~0.5g) approaches emergency
        braking for a laden vehicle.
    smoothing_window : int
        Rolling-mean window (samples) applied to speed_ms before
        differentiation. 3 is the minimum that meaningfully suppresses
        GPS jitter without eroding genuine short braking events.
    min_event_gap_s : float
        Minimum gap between flagged samples before they are treated as
        two separate events rather than one continuous braking action.
        2.0 s absorbs the natural multi-sample tail of a single hard
        stop without merging genuinely distinct events.
    """

    def __init__(
        self,
        decel_threshold_ms2: float = -3.0,
        accel_threshold_ms2: float = 2.5,
        severe_decel_ms2: float = -5.0,
        smoothing_window: int = 3,
        min_event_gap_s: float = 2.0,
    ):
        self.decel_threshold_ms2 = decel_threshold_ms2
        self.accel_threshold_ms2 = accel_threshold_ms2
        self.severe_decel_ms2 = severe_decel_ms2
        self.smoothing_window = smoothing_window
        self.min_event_gap_s = min_event_gap_s

    def _smooth_speed(self, df: pd.DataFrame) -> pd.Series:
        """Rolling-mean smoothing applied before differentiation, never after."""
        return (
            df["speed_ms"]
            .rolling(window=self.smoothing_window, center=True, min_periods=1)
            .mean()
        )

    def compute_acceleration(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Return a copy of df with 'speed_smoothed_ms' and 'accel_ms2' added.

        Expected output shape: same row count; first row has NaN
        acceleration (no predecessor).
        """
        df = df.sort_values("timestamp").copy()
        df["speed_smoothed_ms"] = self._smooth_speed(df)

        dt_s = df["timestamp"].diff().dt.total_seconds()
        dv_ms = df["speed_smoothed_ms"].diff()

        df["accel_ms2"] = dv_ms / dt_s.replace(0, np.nan)
        return df

    def detect_events(self, df: pd.DataFrame) -> List[HarshEvent]:
        """
        Full pipeline: smooth, differentiate, threshold, debounce, grade.

        Parameters
        ----------
        df : DataFrame with 'timestamp' (UTC) and 'speed_ms' for a single
             device / single trip.

        Returns
        -------
        List of HarshEvent, one per debounced braking or acceleration
        incident, ordered by start_time.
        """
        df = self.compute_acceleration(df)

        braking_flag = df["accel_ms2"] <= self.decel_threshold_ms2
        accel_flag = df["accel_ms2"] >= self.accel_threshold_ms2

        events: List[HarshEvent] = []
        events.extend(self._debounce(df, braking_flag, "harsh_braking"))
        events.extend(self._debounce(df, accel_flag, "harsh_acceleration"))
        events.sort(key=lambda e: e.start_time)
        return events

    def _debounce(
        self, df: pd.DataFrame, flag: pd.Series, event_type: str
    ) -> List[HarshEvent]:
        """
        Merge consecutive flagged rows separated by less than
        min_event_gap_s into a single event.
        """
        flagged = df[flag].copy()
        if flagged.empty:
            return []

        gap = flagged["timestamp"].diff().dt.total_seconds()
        new_event = (gap.isna()) | (gap > self.min_event_gap_s)
        flagged["_event_id"] = new_event.cumsum()

        results: List[HarshEvent] = []
        for _, group in flagged.groupby("_event_id"):
            peak = (
                group["accel_ms2"].min()
                if event_type == "harsh_braking"
                else group["accel_ms2"].max()
            )
            severity = (
                "severe"
                if event_type == "harsh_braking" and peak <= self.severe_decel_ms2
                else "moderate"
            )
            results.append(
                HarshEvent(
                    event_type=event_type,
                    start_time=group["timestamp"].iloc[0],
                    end_time=group["timestamp"].iloc[-1],
                    peak_accel_ms2=float(peak),
                    severity=severity,
                    n_samples=len(group),
                )
            )
        return results

Execution & Tuning Guidelines

detector = HarshEventDetector(
    decel_threshold_ms2=-3.0,
    accel_threshold_ms2=2.5,
    smoothing_window=3,
    min_event_gap_s=2.0,
)
events = detector.detect_events(trip_df)
for e in events:
    print(e.event_type, e.severity, e.start_time, round(e.peak_accel_ms2, 2))

Feed events output into a driver-scoring aggregation keyed by vehicle_id and shift, and cross-reference peak timestamps against heading synchronization output — a harsh-braking event that coincides with a sharp heading change is far more likely to indicate a genuine evasive manoeuvre than one on a straight road segment.

Common Pitfalls

Up: Speed Profiling from Raw GPS Coordinates | Trajectory Analysis & Map Matching Techniques

Related

• Speed Profiling from Raw GPS Coordinates — the parent pipeline supplying the smoothed speed series this detector differentiates
• Calculating Instantaneous vs Average Speed from GPS Traces — background on why instantaneous (not average) speed feeds harsh-event detection
• Kalman Filtering for GPS Noise Reduction — an alternative smoothing stage when rolling-mean is insufficient at higher sampling rates
• Directionality & Heading Synchronization — corroborating heading-change data improves confidence that a flagged event is a genuine evasive manoeuvre
• Multi-Modal Route Matching for Mixed Fleets — harsh-event thresholds should be recalibrated per vehicle mode rather than applied uniformly across a mixed fleet