Smooth synthetic sensor data with a moving average and check the delay

01Define the moving average and its timing

This example uses a trailing moving average with a window of three samples. For sample n, the filtered value is the mean of the current sample and the previous two samples: ma[n] = (x[n] + x[n-1] + x[n-2]) / 3. The first two samples therefore do not have a complete three-sample window.

A moving average reduces rapid sample-to-sample variation, but it also changes timing. A three-sample window containing times 0.2, 0.3, and 0.4 seconds is centered at 0.3 seconds even though a causal trailing implementation produces the result when the 0.4-second sample arrives. With a 0.1-second sample interval, this center-to-output alignment difference is 0.1 seconds.

02Create a tiny synthetic acceleration signal

The following dataset is synthetic and was written specifically for this article. Save it as sensor_data.csv. Samples are spaced every 0.1 seconds. The signal begins near 0 and then rises toward 9 m/s^2, with deliberately added variation around both levels.

time_s,accel_m_s2
0.0,0
0.1,3
0.2,-3
0.3,0
0.4,9
0.5,12
0.6,6
0.7,9

There are 8 samples. Before 0.4 seconds, the raw values are 0, 3, -3, and 0. From 0.4 seconds onward, the values are 9, 12, 6, and 9. The data is intentionally simple; it is not claimed to represent a particular vehicle sensor or noise distribution.

03Calculate the filtered values by hand

The first complete window ends at 0.2 seconds: (0 + 3 - 3) / 3 = 0. The next window is (3 - 3 + 0) / 3 = 0. When the raw signal first reaches 9 at 0.4 seconds, the three values in the window are -3, 0, and 9, so the filtered result is 2 rather than 9.

The rise is visibly spread across several outputs: the filtered sequence around the change is 0, 2, 7, 9 instead of jumping directly from 0 to 9. This is the smoothing effect and also illustrates why timing must be considered when filtered signals are compared with other channels.

04Apply the filter and record its alignment delay

Save the following script as moving_average_filter.py. It checks that timestamps increase with a constant 0.1-second interval, calculates a three-sample trailing mean, and writes a new CSV. It refuses to reuse an existing output folder.

import csv
from pathlib import Path

SOURCE = Path("sensor_data.csv")
OUTPUT_DIR = Path("outputs") / "moving_average_result"
OUTPUT = OUTPUT_DIR / "filtered_sensor_data.csv"
WINDOW = 3
EXPECTED_DT = 0.1
TOLERANCE = 1e-9


def main() -> None:
    if not SOURCE.is_file():
        raise FileNotFoundError(f"Source CSV not found: {SOURCE}")
    if OUTPUT_DIR.exists():
        raise FileExistsError(f"Output folder already exists: {OUTPUT_DIR}")
    if WINDOW < 1 or WINDOW % 2 == 0:
        raise ValueError("WINDOW must be a positive odd integer.")

    samples = []
    with SOURCE.open("r", encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream)
        required = {"time_s", "accel_m_s2"}
        if reader.fieldnames is None or not required.issubset(reader.fieldnames):
            raise ValueError("CSV is missing a required column.")

        for row in reader:
            samples.append((float(row["time_s"]), float(row["accel_m_s2"])))

    if len(samples) < WINDOW:
        raise ValueError("Not enough samples for the moving-average window.")

    for index in range(1, len(samples)):
        dt = samples[index][0] - samples[index - 1][0]
        if abs(dt - EXPECTED_DT) > TOLERANCE:
            raise ValueError(f"Unexpected sample interval at row {index + 2}: {dt}")

    half_window = WINDOW // 2
    output_rows = []

    for index, (time_s, raw_value) in enumerate(samples):
        if index < WINDOW - 1:
            average_text = ""
            center_text = ""
            delay_text = ""
        else:
            window_values = [value for _, value in samples[index - WINDOW + 1:index + 1]]
            average = sum(window_values) / WINDOW
            center_time = samples[index - half_window][0]
            alignment_delay = time_s - center_time
            average_text = f"{average:.3f}"
            center_text = f"{center_time:.1f}"
            delay_text = f"{alignment_delay:.1f}"

        output_rows.append({
            "time_s": f"{time_s:.1f}",
            "raw_accel_m_s2": f"{raw_value:g}",
            "ma3_accel_m_s2": average_text,
            "window_center_time_s": center_text,
            "alignment_delay_s": delay_text,
        })

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()
    fields = [
        "time_s", "raw_accel_m_s2", "ma3_accel_m_s2",
        "window_center_time_s", "alignment_delay_s"
    ]
    with OUTPUT.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=fields)
        writer.writeheader()
        writer.writerows(output_rows)

    valid_outputs = len(samples) - WINDOW + 1
    print(f"Input samples: {len(samples)}.")
    print(f"MA{WINDOW} outputs with full windows: {valid_outputs}.")
    print(f"Nominal alignment delay: {half_window * EXPECTED_DT:.1f} s.")
    print(f"Output: {OUTPUT.as_posix()}")


if __name__ == "__main__":
    main()

05Compare the expected result

The first two filtered cells should be blank because a complete three-sample window is not available. The remaining six filtered values should be 0.000, 0.000, 2.000, 7.000, 9.000, and 9.000.

time_s,raw_accel_m_s2,ma3_accel_m_s2,window_center_time_s,alignment_delay_s
0.0,0,,,
0.1,3,,,
0.2,-3,0.000,0.1,0.1
0.3,0,0.000,0.2,0.1
0.4,9,2.000,0.3,0.1
0.5,12,7.000,0.4,0.1
0.6,6,9.000,0.5,0.1
0.7,9,9.000,0.6,0.1

The expected console output below was derived by hand from the synthetic data and code. It is not a captured execution log.

Input samples: 8.
MA3 outputs with full windows: 6.
Nominal alignment delay: 0.1 s.
Output: outputs/moving_average_result/filtered_sensor_data.csv

06Check the result and common errors

• Verify the first complete average manually: (0 + 3 - 3) / 3 = 0.
• Verify the transition value at 0.4 seconds: (-3 + 0 + 9) / 3 = 2.
• Confirm that the filter produces 6 full-window values from 8 samples with a window of 3.
• Confirm that every reported window-center difference is 0.1 seconds.
• Run the script again without changing OUTPUT_DIR. It should stop with FileExistsError instead of overwriting the result.

07Understand the limits before using test data

A moving average is a simple low-pass smoothing method, not a universal sensor filter. It can reduce rapid fluctuations, but it can also attenuate peaks and blur the start or end of short events. That matters in vibration, impact, fault-detection, and threshold-based analyses.

The 0.1-second delay reported here is the difference between the trailing window's center time and the timestamp at which its result becomes available for a three-sample window sampled every 0.1 seconds. It should not be treated as a universal delay for every signal feature or every filtering implementation.

Real automotive data may have nonuniform sampling, dropped samples, sensor bias, saturation, multiple synchronized channels, or frequency content that requires a filter designed from sampling rate and bandwidth requirements. Always preserve the raw signal and document the filtering method before drawing engineering conclusions.

Execution and verification record

ChatGPT Python tool · Python 3.x · standard library: csv, pathlib

• Execution completed successfully with 8 input samples and 6 full-window MA3 outputs.
• Generated MA3 values were exactly 0.000, 0.000, 2.000, 7.000, 9.000, and 9.000.
• Generated window-center times were 0.1, 0.2, 0.3, 0.4, 0.5, and 0.6 s, with 0.1 s alignment delay for every complete window.
• The generated filtered_sensor_data.csv matched the article's expected CSV exactly.
• Console output matched the claimed values and output path.
• Second execution raised FileExistsError for outputs/moving_average_result, confirming overwrite protection.
• SHA-256 comparison confirmed sensor_data.csv was unchanged.

References

The explanations and examples were written for this site. See the official sources below for the related behavior and concepts.

• Official Python 3.12 documentation · csv ↗ (New tab)
• Official Python 3.12 documentation · pathlib ↗ (New tab)