移動平均で合成センサーデータを平滑化し、遅延を確認する

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

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

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

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