意味・単位・型・許容値を含むデータ辞書を作成する

run_id,speed_kmh,accel_m_s2,road_surface,test_status
R001,30,0.5,dry,complete
R002,50,-1.2,wet,complete
R003,70,0.0,dry,review
R004,40,-0.8,wet,complete
R005,135,fast,snow,done

import csv
import re
from pathlib import Path

SOURCE = Path("test_runs.csv")
OUTPUT_DIR = Path("outputs") / "data_dictionary_result"
DICTIONARY = OUTPUT_DIR / "data_dictionary.csv"
ISSUES = OUTPUT_DIR / "validation_issues.csv"

SPEC = [
    {
        "column_name": "run_id",
        "meaning": "Unique identifier for one test run",
        "unit": "",
        "type": "string",
        "allowed_values": "R followed by exactly 3 digits; unique",
        "example": "R001",
    },
    {
        "column_name": "speed_kmh",
        "meaning": "Vehicle speed at the start of the record",
        "unit": "km/h",
        "type": "integer",
        "allowed_values": "0 to 120 inclusive",
        "example": "50",
    },
    {
        "column_name": "accel_m_s2",
        "meaning": "Longitudinal acceleration",
        "unit": "m/s^2",
        "type": "float",
        "allowed_values": "-5.0 to 5.0 inclusive",
        "example": "-1.2",
    },
    {
        "column_name": "road_surface",
        "meaning": "Road-surface condition used for the test",
        "unit": "",
        "type": "category",
        "allowed_values": "dry;wet",
        "example": "wet",
    },
    {
        "column_name": "test_status",
        "meaning": "Review state of the test record",
        "unit": "",
        "type": "category",
        "allowed_values": "complete;review",
        "example": "complete",
    },
]


def issue(row_number, column, value, problem):
    return {
        "row_number": row_number,
        "column_name": column,
        "value": value,
        "issue": problem,
    }


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

    expected_columns = [item["column_name"] for item in SPEC]
    issues = []
    seen_run_ids = set()
    row_count = 0

    with SOURCE.open("r", encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream)
        if reader.fieldnames != expected_columns:
            raise ValueError(
                f"Header mismatch. Expected {expected_columns}, got {reader.fieldnames}."
            )

        for row_number, row in enumerate(reader, start=2):
            row_count += 1

            run_id = row["run_id"].strip()
            if re.fullmatch(r"R[0-9]{3}", run_id) is None:
                issues.append(issue(row_number, "run_id", run_id, "PATTERN_ERROR"))
            elif run_id in seen_run_ids:
                issues.append(issue(row_number, "run_id", run_id, "DUPLICATE_VALUE"))
            seen_run_ids.add(run_id)

            try:
                speed = int(row["speed_kmh"])
            except ValueError:
                issues.append(issue(row_number, "speed_kmh", row["speed_kmh"], "TYPE_ERROR"))
            else:
                if not 0 <= speed <= 120:
                    issues.append(issue(row_number, "speed_kmh", row["speed_kmh"], "OUT_OF_RANGE"))

            try:
                accel = float(row["accel_m_s2"])
            except ValueError:
                issues.append(issue(row_number, "accel_m_s2", row["accel_m_s2"], "TYPE_ERROR"))
            else:
                if not -5.0 <= accel <= 5.0:
                    issues.append(issue(row_number, "accel_m_s2", row["accel_m_s2"], "OUT_OF_RANGE"))

            if row["road_surface"] not in {"dry", "wet"}:
                issues.append(issue(row_number, "road_surface", row["road_surface"], "ALLOWED_VALUE_ERROR"))

            if row["test_status"] not in {"complete", "review"}:
                issues.append(issue(row_number, "test_status", row["test_status"], "ALLOWED_VALUE_ERROR"))

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()

    dictionary_fields = [
        "column_name", "meaning", "unit", "type", "allowed_values", "example"
    ]
    with DICTIONARY.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=dictionary_fields)
        writer.writeheader()
        writer.writerows(SPEC)

    issue_fields = ["row_number", "column_name", "value", "issue"]
    with ISSUES.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=issue_fields)
        writer.writeheader()
        writer.writerows(issues)

    print(f"Rows checked: {row_count}.")
    print(f"Dictionary columns: {len(SPEC)}.")
    print(f"Validation issues: {len(issues)}.")
    print(f"Output folder: {OUTPUT_DIR.as_posix()}")


if __name__ == "__main__":
    main()

row_number,column_name,value,issue
6,speed_kmh,135,OUT_OF_RANGE
6,accel_m_s2,fast,TYPE_ERROR
6,road_surface,snow,ALLOWED_VALUE_ERROR
6,test_status,done,ALLOWED_VALUE_ERROR

Rows checked: 5.
Dictionary columns: 5.
Validation issues: 4.
Output folder: outputs/data_dictionary_result