Suavizar datos sintéticos de sensores con una media móvil y comprobar el retraso

01Definir la media móvil y su relación temporal

Este ejemplo utiliza una media móvil retrospectiva con una ventana de tres muestras. Para la muestra n, el valor filtrado es la media de la muestra actual y las dos muestras anteriores: ma[n] = (x[n] + x[n-1] + x[n-2]) / 3. Por lo tanto, las dos primeras muestras no disponen de una ventana completa de tres muestras.

Una media móvil reduce las variaciones rápidas entre muestras, pero también modifica la relación temporal. Una ventana de tres muestras que contiene los tiempos 0.2, 0.3 y 0.4 segundos está centrada en 0.3 segundos, aunque una implementación causal retrospectiva produce el resultado cuando llega la muestra de 0.4 segundos. Con un intervalo de muestreo de 0.1 segundos, esta diferencia de alineación entre el centro y la salida es de 0.1 segundos.

02Crear una pequeña señal sintética de aceleración

El siguiente conjunto de datos es sintético y se creó específicamente para este artículo. Guárdalo como sensor_data.csv. Las muestras están separadas cada 0.1 segundos. La señal comienza cerca de 0 y después aumenta hacia 9 m/s^2, con variaciones añadidas deliberadamente alrededor de ambos niveles.

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

Hay 8 muestras. Antes de 0.4 segundos, los valores sin procesar son 0, 3, -3 y 0. A partir de 0.4 segundos, los valores son 9, 12, 6 y 9. Los datos son intencionadamente simples; no se afirma que representen un sensor concreto de un vehículo ni una distribución específica de ruido.

03Calcular manualmente los valores filtrados

La primera ventana completa termina en 0.2 segundos: (0 + 3 - 3) / 3 = 0. La siguiente ventana es (3 - 3 + 0) / 3 = 0. Cuando la señal sin procesar alcanza por primera vez 9 en 0.4 segundos, los tres valores de la ventana son -3, 0 y 9, por lo que el resultado filtrado es 2 en lugar de 9.

El ascenso se distribuye visiblemente entre varias salidas: la secuencia filtrada alrededor del cambio es 0, 2, 7, 9 en lugar de saltar directamente de 0 a 9. Este es el efecto de suavizado y también muestra por qué debe tenerse en cuenta la sincronización al comparar señales filtradas con otros canales.

04Aplicar el filtro y registrar su retraso de alineación

Guarda el siguiente script como moving_average_filter.py. Comprueba que las marcas de tiempo aumenten con un intervalo constante de 0.1 segundos, calcula una media retrospectiva de tres muestras y escribe un nuevo CSV. Se niega a reutilizar una carpeta de salida existente.

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()

05Comparar con el resultado esperado

Las dos primeras celdas filtradas deben quedar vacías porque no hay disponible una ventana completa de tres muestras. Los seis valores filtrados restantes deben ser 0.000, 0.000, 2.000, 7.000, 9.000 y 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

La salida de consola esperada que aparece a continuación se obtuvo manualmente a partir de los datos sintéticos y del código. No es un registro de ejecución capturado.

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

06Comprobar el resultado y los errores habituales

• Verifica manualmente la primera media completa: (0 + 3 - 3) / 3 = 0.
• Verifica el valor de transición en 0.4 segundos: (-3 + 0 + 9) / 3 = 2.
• Confirma que el filtro produce 6 valores de ventana completa a partir de 8 muestras con una ventana de 3.
• Confirma que todas las diferencias informadas respecto al centro de la ventana son de 0.1 segundos.
• Ejecuta de nuevo el script sin cambiar OUTPUT_DIR. Debe detenerse con FileExistsError en lugar de sobrescribir el resultado.

07Entender los límites antes de usar datos de ensayo

Una media móvil es un método sencillo de suavizado de paso bajo, no un filtro universal para sensores. Puede reducir fluctuaciones rápidas, pero también puede atenuar picos y difuminar el inicio o el final de eventos breves. Esto es importante en análisis de vibración, impacto, detección de fallos y análisis basados en umbrales.

El retraso de 0.1 segundos indicado aquí es la diferencia entre el tiempo central de la ventana retrospectiva y la marca de tiempo en la que su resultado pasa a estar disponible para una ventana de tres muestras tomadas cada 0.1 segundos. No debe considerarse un retraso universal para cualquier característica de señal o cualquier implementación de filtrado.

Los datos automotrices reales pueden presentar muestreo no uniforme, muestras perdidas, sesgo del sensor, saturación, varios canales sincronizados o contenido frecuencial que requiera un filtro diseñado a partir de la frecuencia de muestreo y los requisitos de ancho de banda. Conserva siempre la señal sin procesar y documenta el método de filtrado antes de extraer conclusiones de ingeniería.

Registro de ejecución y verificación

2026-09-20 · ejemplo comprobado manualmente · objetivo: Python 3.12 · biblioteca estándar: csv, pathlib · sin ejecución

• Se comprobaron manualmente las 8 muestras sintéticas de aceleración y su separación constante de 0.1 segundos.
• Se calcularon manualmente las seis medias completas de tres muestras como 0, 0, 2, 7, 9 y 9.
• Se comprobó manualmente que una ventana de tres muestras tiene su centro una muestra antes de la marca temporal de salida retrospectiva.
• Se calculó la diferencia nominal de alineación entre el centro y la salida como 1 sample × 0.1 s = 0.1 s.
• Se inspeccionó el script para comprobar las columnas de entrada, los intervalos de muestreo, la protección frente a colisiones de salida y la conservación del CSV original.
• Se derivaron manualmente el CSV de salida esperado y el texto esperado de consola.

Criterios de redacción y verificación de todo el sitio

Fuentes de referencia

Las explicaciones y los ejemplos son de elaboración propia. Puedes consultar los comportamientos y conceptos relacionados en las siguientes fuentes oficiales.

• Documentación oficial de Python 3.12 · csv ↗ (New tab)
• Documentación oficial de Python 3.12 · pathlib ↗ (New tab)