自動車試験データを km/h、m/s、g、m/s² の間で変換する

sample_id,speed_kmh,accel_g
T001,0,0
T002,36,1
T003,72,-0.5
T004,90,0.25
T005,108,-1.25

import csv
import math
from pathlib import Path

SOURCE = Path("unit_test_data.csv")
OUTPUT_DIR = Path("outputs") / "unit_conversion_result"
OUTPUT = OUTPUT_DIR / "converted_test_data.csv"
STANDARD_GRAVITY = 9.80665
TOLERANCE = 1e-12


def kmh_to_ms(value: float) -> float:
    return value / 3.6


def ms_to_kmh(value: float) -> float:
    return value * 3.6


def g_to_ms2(value: float) -> float:
    return value * STANDARD_GRAVITY


def ms2_to_g(value: float) -> float:
    return value / STANDARD_GRAVITY


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}")

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

        seen_ids = set()
        for row in reader:
            sample_id = row["sample_id"].strip()
            if not sample_id:
                raise ValueError("sample_id must not be blank.")
            if sample_id in seen_ids:
                raise ValueError(f"Duplicate sample_id: {sample_id}")
            seen_ids.add(sample_id)

            speed_kmh = float(row["speed_kmh"])
            accel_g = float(row["accel_g"])
            speed_ms = kmh_to_ms(speed_kmh)
            accel_ms2 = g_to_ms2(accel_g)
            speed_roundtrip = ms_to_kmh(speed_ms)
            accel_roundtrip = ms2_to_g(accel_ms2)

            speed_ok = math.isclose(
                speed_kmh, speed_roundtrip,
                rel_tol=TOLERANCE, abs_tol=TOLERANCE
            )
            accel_ok = math.isclose(
                accel_g, accel_roundtrip,
                rel_tol=TOLERANCE, abs_tol=TOLERANCE
            )
            if not speed_ok or not accel_ok:
                raise RuntimeError(f"Round-trip check failed: {sample_id}")

            output_rows.append({
                "sample_id": sample_id,
                "speed_kmh": f"{speed_kmh:g}",
                "speed_m_s": f"{speed_ms:.6f}",
                "speed_roundtrip_kmh": f"{speed_roundtrip:.6f}",
                "accel_g": f"{accel_g:g}",
                "accel_m_s2": f"{accel_ms2:.6f}",
                "accel_roundtrip_g": f"{accel_roundtrip:.6f}",
                "roundtrip_check": "PASS",
            })

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()
    fields = [
        "sample_id", "speed_kmh", "speed_m_s", "speed_roundtrip_kmh",
        "accel_g", "accel_m_s2", "accel_roundtrip_g", "roundtrip_check"
    ]
    with OUTPUT.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=fields)
        writer.writeheader()
        writer.writerows(output_rows)

    print(f"Converted records: {len(output_rows)}.")
    print(f"Round-trip checks passed: {len(output_rows)}.")
    print(f"Standard gravity used: {STANDARD_GRAVITY} m/s^2.")
    print(f"Output: {OUTPUT.as_posix()}")


if __name__ == "__main__":
    main()

sample_id,speed_kmh,speed_m_s,speed_roundtrip_kmh,accel_g,accel_m_s2,accel_roundtrip_g,roundtrip_check
T001,0,0.000000,0.000000,0,0.000000,0.000000,PASS
T002,36,10.000000,36.000000,1,9.806650,1.000000,PASS
T003,72,20.000000,72.000000,-0.5,-4.903325,-0.500000,PASS
T004,90,25.000000,90.000000,0.25,2.451663,0.250000,PASS
T005,108,30.000000,108.000000,-1.25,-12.258313,-1.250000,PASS

Converted records: 5.
Round-trip checks passed: 5.
Standard gravity used: 9.80665 m/s^2.
Output: outputs/unit_conversion_result/converted_test_data.csv