tdkpin/original/tools/promote_verified_data.py

#!/usr/bin/env python3
"""Promote semantic DGROUP rows after every referring function is verified."""

from __future__ import annotations

import argparse
import csv
import re
from io import StringIO
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent


def closed_evidence(text: str) -> str:
    """Remove obsolete pre-closure clauses without discarding useful evidence."""
    stale = re.compile(
        r"\b(pending|remain(?:s|ing)? raw|raw/partial|remain(?:s|ing)? partial|"
        r"unresolved)\b",
        re.IGNORECASE,
    )
    clauses = [
        clause.strip()
        for clause in text.split(";")
        if clause.strip() and not stale.search(clause)
    ]
    closure = "complete verified referring-function closure"
    if not any(closure in clause for clause in clauses):
        clauses.append(closure)
    return "; ".join(clauses)


def read_tsv(name: str) -> list[dict[str, str]]:
    with (ROOT / name).open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle, delimiter="\t"))


def serialize(columns: list[str], rows: list[dict[str, str]]) -> str:
    output = StringIO()
    writer = csv.DictWriter(
        output, fieldnames=columns, delimiter="\t", lineterminator="\n"
    )
    writer.writeheader()
    writer.writerows(rows)
    return output.getvalue()


def promote() -> dict[str, str]:
    functions = read_tsv("FUNCTION_RECONSTRUCTION.tsv")
    verified_names = {
        row["raw_name"] for row in functions if row["status"] == "verified"
    }
    if len(verified_names) != len(functions):
        raise SystemExit("refusing data promotion before all functions are verified")

    reconstruction = read_tsv("DATA_RECONSTRUCTION.tsv")
    if any(row["status"] == "unknown" for row in reconstruction):
        raise SystemExit("refusing data promotion while referenced addresses are unknown")
    for row in reconstruction:
        referring = {
            name for name in row["referring_functions"].split(",") if name
        }
        unresolved = referring - verified_names
        if unresolved:
            raise SystemExit(
                f"unverified data callers at {row['address']}: "
                + ", ".join(sorted(unresolved))
            )
        row["evidence"] = closed_evidence(row["evidence"])
        if row["status"] == "restored":
            row["status"] = "verified"
            row["notes"] = (
                "Exact semantic object field, extent, and complete read/write "
                "inventory are closed through verified address-linked functions."
            )

    objects = read_tsv("DATA_OBJECTS.tsv")
    for row in objects:
        row["evidence"] = closed_evidence(row["evidence"])
        if row["status"] == "restored":
            row["status"] = "verified"
            row["notes"] = (
                "Exact object boundary, type, initialization, and full in-image "
                "reader/writer lifecycle are verified."
            )

    coverage = read_tsv("DATA_COVERAGE.tsv")
    for row in coverage:
        row["evidence"] = closed_evidence(row["evidence"])
        if row["status"] == "restored":
            row["status"] = "verified"
            row["notes"] = (
                "Range is fully covered by a verified semantic DGROUP object."
            )

    return {
        "DATA_RECONSTRUCTION.tsv": serialize(
            list(reconstruction[0]), reconstruction
        ),
        "DATA_OBJECTS.tsv": serialize(list(objects[0]), objects),
        "DATA_COVERAGE.tsv": serialize(list(coverage[0]), coverage),
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--check", action="store_true")
    arguments = parser.parse_args()
    expected = promote()
    for name, text in expected.items():
        path = ROOT / name
        if arguments.check:
            if path.read_text(encoding="utf-8") != text:
                raise SystemExit(f"{name} needs deterministic semantic promotion")
        else:
            path.write_text(text, encoding="utf-8")


if __name__ == "__main__":
    main()