marte-benchmark-suite/scenarios/S03H1_execution_profiles/export_s03_csv.py

#!/usr/bin/env python3
"""Export the approved S03 binary record layout to a derived CSV artifact."""

import argparse
import csv
import hashlib
import os
import struct
import sys
import tempfile
from pathlib import Path


UINT32_TYPE_HEX = "0408"
SIGNAL_HEADER_BYTES = 38
EXPECTED_SIGNALS = (
    ("Counter", 1),
    ("Time", 1),
    ("State1_Thread1_CycleTime", 1),
    ("SentinelScalarA", 1),
    ("SentinelScalarB", 1),
    ("SentinelVector", 3),
)
EXPECTED_HEADER_BYTES = 4 + len(EXPECTED_SIGNALS) * SIGNAL_HEADER_BYTES
EXPECTED_RECORD_BYTES = 32
CSV_COLUMNS = (
    "Counter",
    "Time",
    "CycleTime",
    "SentinelScalarA",
    "SentinelScalarB",
    "SentinelVector0",
    "SentinelVector1",
    "SentinelVector2",
)


class ExportError(Exception):
    """A fail-closed binary or publication error."""


def decode_name(raw):
    try:
        return raw.split(b"\x00", 1)[0].decode("ascii")
    except UnicodeDecodeError as exc:
        raise ExportError("NON_ASCII_SIGNAL_NAME") from exc


def read_records(binary_path):
    try:
        data = binary_path.read_bytes()
    except OSError as exc:
        raise ExportError(f"BINARY_READ_FAILED:{exc}") from exc

    if len(data) < EXPECTED_HEADER_BYTES:
        raise ExportError(
            f"HEADER_TOO_SMALL:EXPECTED_{EXPECTED_HEADER_BYTES}:OBSERVED_{len(data)}"
        )

    signal_count = struct.unpack_from("<I", data, 0)[0]
    if signal_count != len(EXPECTED_SIGNALS):
        raise ExportError(
            f"SIGNAL_COUNT_MISMATCH:EXPECTED_{len(EXPECTED_SIGNALS)}:OBSERVED_{signal_count}"
        )

    offset = 4
    observed = []
    for _ in range(signal_count):
        type_hex = data[offset : offset + 2].hex()
        offset += 2
        name = decode_name(data[offset : offset + 32])
        offset += 32
        elements = struct.unpack_from("<I", data, offset)[0]
        offset += 4
        observed.append((name, type_hex, elements))

    expected = tuple(
        (name, UINT32_TYPE_HEX, elements) for name, elements in EXPECTED_SIGNALS
    )
    if tuple(observed) != expected:
        raise ExportError(
            "HEADER_CONTRACT_MISMATCH:"
            f"EXPECTED_{expected!r}:OBSERVED_{tuple(observed)!r}"
        )
    if offset != EXPECTED_HEADER_BYTES:
        raise ExportError(
            f"HEADER_SIZE_MISMATCH:EXPECTED_{EXPECTED_HEADER_BYTES}:OBSERVED_{offset}"
        )

    payload = data[offset:]
    remainder = len(payload) % EXPECTED_RECORD_BYTES
    if remainder != 0:
        raise ExportError(
            f"PAYLOAD_ALIGNMENT_ERROR:REMAINDER_{remainder}:RECORD_{EXPECTED_RECORD_BYTES}"
        )
    if not payload:
        raise ExportError("NO_COMPLETE_RECORDS")

    records = [
        struct.unpack_from("<8I", payload, record_offset)
        for record_offset in range(0, len(payload), EXPECTED_RECORD_BYTES)
    ]
    return data, records


def write_csv_atomic(output_path, records):
    output_path.parent.mkdir(parents=True, exist_ok=True)
    if output_path.exists() and not output_path.is_file():
        raise ExportError(f"OUTPUT_NOT_REGULAR_FILE:{output_path}")

    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{output_path.name}.",
        suffix=".tmp",
        dir=output_path.parent,
        text=True,
    )
    temporary_path = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as stream:
            writer = csv.writer(stream, lineterminator="\n")
            writer.writerow(CSV_COLUMNS)
            writer.writerows(records)
            stream.flush()
            os.fsync(stream.fileno())

        if output_path.exists():
            if temporary_path.read_bytes() == output_path.read_bytes():
                temporary_path.unlink()
                return "ALREADY_VALID"
            raise ExportError(f"OUTPUT_EXISTS_BUT_DIFFERS:{output_path}")

        os.chmod(temporary_path, 0o644)
        os.replace(temporary_path, output_path)
        return "CREATED"
    finally:
        if temporary_path.exists():
            temporary_path.unlink()


def sha256_bytes(data):
    return hashlib.sha256(data).hexdigest()


def main():
    parser = argparse.ArgumentParser(
        description=(
            "Create a human-readable CSV from the S03 FileWriter binary. "
            "The binary remains the primary evidence."
        )
    )
    parser.add_argument("binary", type=Path)
    parser.add_argument("csv_output", type=Path, nargs="?")
    args = parser.parse_args()

    output_path = args.csv_output
    if output_path is None:
        output_path = args.binary.with_suffix(".csv")

    try:
        binary_data, records = read_records(args.binary)
        publication = write_csv_atomic(output_path, records)
        csv_data = output_path.read_bytes()
    except ExportError as exc:
        print("CSV_EXPORT_STATUS=FAIL")
        print(f"CSV_EXPORT_ERROR={exc}")
        return 40
    except OSError as exc:
        print("CSV_EXPORT_STATUS=FAIL")
        print(f"CSV_EXPORT_ERROR=OS_ERROR:{exc}")
        return 41

    print("PRIMARY_EVIDENCE=s03_output.bin")
    print("DERIVED_ARTIFACT=s03_output.csv")
    print(f"BINARY_PATH={args.binary}")
    print(f"BINARY_SHA256={sha256_bytes(binary_data)}")
    print(f"CSV_PATH={output_path}")
    print(f"CSV_SHA256={sha256_bytes(csv_data)}")
    print(f"CSV_RECORD_COUNT={len(records)}")
    print(f"CSV_PUBLICATION={publication}")
    print("CSV_EXPORT_STATUS=PASS")
    return 0


if __name__ == "__main__":
    sys.exit(main())