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

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

• A execução foi concluída com sucesso com 8 amostras de entrada e 6 saídas MA3 de janela completa.
• Os valores MA3 gerados foram exatamente 0.000, 0.000, 2.000, 7.000, 9.000 e 9.000.
• Os tempos centrais de janela gerados foram 0.1, 0.2, 0.3, 0.4, 0.5 e 0.6 s, com atraso de alinhamento de 0.1 s para cada janela completa.
• O filtered_sensor_data.csv gerado correspondeu exatamente ao CSV esperado do artigo.
• A saída do console correspondeu aos valores e ao caminho de saída declarados.
• A segunda execução gerou FileExistsError para outputs/moving_average_result, confirmando a proteção contra sobrescrita.
• A comparação SHA-256 confirmou que sensor_data.csv permaneceu inalterado.

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)