Suavize dados sintéticos de sensores com média móvel e verifique o atraso

01Defina a média móvel e sua relação temporal

Este exemplo usa uma média móvel retrospectiva com uma janela de três amostras. Para a amostra n, o valor filtrado é a média da amostra atual e das duas amostras anteriores: ma[n] = (x[n] + x[n-1] + x[n-2]) / 3. Portanto, as duas primeiras amostras não têm uma janela completa de três amostras.

Uma média móvel reduz variações rápidas entre amostras, mas também altera a relação temporal. Uma janela de três amostras contendo os tempos 0.2, 0.3 e 0.4 segundos é centrada em 0.3 segundos, embora uma implementação causal retrospectiva produza o resultado quando a amostra de 0.4 segundos chega. Com um intervalo de amostragem de 0.1 segundos, essa diferença de alinhamento entre o centro e a saída é de 0.1 segundos.

02Crie um pequeno sinal sintético de aceleração

O conjunto de dados a seguir é sintético e foi criado especificamente para este artigo. Salve-o como sensor_data.csv. As amostras estão espaçadas a cada 0.1 segundos. O sinal começa próximo de 0 e depois sobe em direção a 9 m/s^2, com variações adicionadas deliberadamente em torno dos dois níveis.

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

Há 8 amostras. Antes de 0.4 segundos, os valores brutos são 0, 3, -3 e 0. A partir de 0.4 segundos, os valores são 9, 12, 6 e 9. Os dados são intencionalmente simples; não se afirma que representem um sensor específico de veículo nem uma distribuição específica de ruído.

03Calcule manualmente os valores filtrados

A primeira janela completa termina em 0.2 segundos: (0 + 3 - 3) / 3 = 0. A janela seguinte é (3 - 3 + 0) / 3 = 0. Quando o sinal bruto chega pela primeira vez a 9 em 0.4 segundos, os três valores da janela são -3, 0 e 9, portanto o resultado filtrado é 2 em vez de 9.

A subida fica visivelmente distribuída por várias saídas: a sequência filtrada em torno da mudança é 0, 2, 7, 9, em vez de saltar diretamente de 0 para 9. Esse é o efeito de suavização e também mostra por que a temporização deve ser considerada ao comparar sinais filtrados com outros canais.

04Aplique o filtro e registre seu atraso de alinhamento

Salve o script a seguir como moving_average_filter.py. Ele verifica se os timestamps aumentam com um intervalo constante de 0.1 segundos, calcula uma média retrospectiva de três amostras e grava um novo CSV. Ele se recusa a reutilizar uma pasta de saída 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()

05Compare com o resultado esperado

As duas primeiras células filtradas devem ficar em branco porque não há uma janela completa de três amostras disponível. Os seis valores filtrados restantes devem ser 0.000, 0.000, 2.000, 7.000, 9.000 e 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

A saída esperada do console abaixo foi derivada manualmente a partir dos dados sintéticos e do código. Não é um log de execução 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

06Verifique o resultado e os erros comuns

• Verifique manualmente a primeira média completa: (0 + 3 - 3) / 3 = 0.
• Verifique o valor de transição em 0.4 segundos: (-3 + 0 + 9) / 3 = 2.
• Confirme que o filtro produz 6 valores de janela completa a partir de 8 amostras com uma janela de 3.
• Confirme que todas as diferenças informadas em relação ao centro da janela são de 0.1 segundos.
• Execute o script novamente sem alterar OUTPUT_DIR. Ele deve parar com FileExistsError em vez de sobrescrever o resultado.

07Entenda os limites antes de usar dados de teste

Uma média móvel é um método simples de suavização passa-baixa, não um filtro universal para sensores. Ela pode reduzir flutuações rápidas, mas também pode atenuar picos e suavizar o início ou o fim de eventos curtos. Isso é relevante em análises de vibração, impacto, detecção de falhas e análises baseadas em limiares.

O atraso de 0.1 segundos informado aqui é a diferença entre o tempo central da janela retrospectiva e o timestamp em que seu resultado se torna disponível para uma janela de três amostras coletadas a cada 0.1 segundos. Ele não deve ser tratado como um atraso universal para toda característica de sinal ou toda implementação de filtragem.

Dados automotivos reais podem apresentar amostragem não uniforme, amostras perdidas, viés do sensor, saturação, vários canais sincronizados ou conteúdo de frequência que exija um filtro projetado a partir da taxa de amostragem e dos requisitos de largura de banda. Preserve sempre o sinal bruto e documente o método de filtragem antes de tirar conclusões de engenharia.

Registro de execução e verificação

2026-09-20 · exemplo verificado manualmente · alvo: Python 3.12 · biblioteca padrão: csv, pathlib · sem execução

• Foram verificadas manualmente as 8 amostras sintéticas de aceleração e seu espaçamento constante de 0.1 segundos.
• Foram calculadas manualmente as seis médias completas de três amostras como 0, 0, 2, 7, 9 e 9.
• Foi verificado manualmente que uma janela de três amostras tem seu centro uma amostra antes do timestamp de saída retrospectiva.
• A diferença nominal de alinhamento entre o centro e a saída foi calculada como 1 sample × 0.1 s = 0.1 s.
• O script foi inspecionado quanto às verificações das colunas de entrada, verificações do intervalo de amostragem, proteção contra colisões de saída e preservação do CSV original.
• O CSV de saída esperado e o texto esperado do console foram derivados manualmente.

Princípios de redação e verificação de todo o site

Fontes de referência

As explicações e os exemplos são de elaboração própria. Os comportamentos e conceitos relacionados podem ser consultados nas fontes oficiais abaixo.

• Documentação oficial do Python 3.12 · csv ↗ (New tab)
• Documentação oficial do Python 3.12 · pathlib ↗ (New tab)