scripts/make_paper_artifacts.py

#!/usr/bin/env python
from __future__ import annotations

import argparse
import csv
import json
import math
import struct
import sys
import zlib
from collections import OrderedDict
from collections.abc import Iterable
from pathlib import Path
from typing import Any

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from dovla_cil.utils.io import ensure_dir, write_json  # noqa: E402, I001


METRIC_ALIASES = OrderedDict(
    [
        (
            "success_rate",
            (
                "task_success_rate",
                "selected_success_rate",
                "success_rate",
                "eval.task_success_rate",
            ),
        ),
        (
            "ranking_acc",
            ("pairwise_ranking_accuracy", "ranking_acc", "eval.pairwise_ranking_accuracy"),
        ),
        ("top1_action_selection", ("top1_action_selection", "eval.top1_action_selection")),
        (
            "instruction_switch_acc",
            (
                "instruction_switch_accuracy",
                "instruction_switch_acc",
                "eval.instruction_switch_accuracy",
            ),
        ),
        ("effect_mae", ("effect_prediction_mae", "effect_mae", "eval.effect_prediction_mae")),
        (
            "regret_ece",
            ("regret_calibration_error", "regret_ece", "eval.regret_calibration_error"),
        ),
        ("ndcg_at_k", ("ndcg_at_k", "eval.ndcg_at_k")),
    ]
)

BASELINE_NAMES = {
    "expert_only_bc",
    "more_independent_demos",
    "random_negatives",
    "cross_state_negatives",
    "label_only_counterfactual",
    "world_model_auxiliary",
    "no_effect_head",
    "no_rank_regret",
}

ABLATION_NAMES = {"world_model_auxiliary", "no_effect_head", "no_rank_regret"}


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Collect DoVLA-CIL runs and create paper-ready tables, figures, and summary."
    )
    parser.add_argument("--runs", type=Path, default=Path("runs"), help="Run directory to scan.")
    parser.add_argument(
        "--out", type=Path, default=Path("paper_artifacts"), help="Output directory."
    )
    args = parser.parse_args(argv)

    summary = generate_paper_artifacts(args.runs, args.out)
    print(f"paper artifacts: {args.out}")
    print(f"result summary: {summary['result_summary']}")
    print(f"tables: {len(summary['tables'])}")
    print(f"figures: {len(summary['figures'])}")
    return 0


def generate_paper_artifacts(runs_dir: str | Path, out_dir: str | Path) -> dict[str, Any]:
    runs_path = Path(runs_dir)
    output_dir = ensure_dir(out_dir)
    figures_dir = ensure_dir(output_dir / "figures")

    payloads = load_result_payloads(runs_path)
    metric_rows = collect_metric_rows(payloads, runs_path)
    scaling_rows = collect_scaling_rows(payloads, runs_path)
    baseline_rows = collect_baseline_rows(metric_rows)
    ablation_rows = [row for row in baseline_rows if row.get("baseline") in ABLATION_NAMES]
    category_rows = collect_per_category_rows(payloads)

    table_paths = {
        "main_scaling": write_table_pair(
            output_dir / "main_scaling_table",
            scaling_rows,
            [
                "run_name",
                "k",
                "num_states",
                "effective_total_records",
                "success_rate",
                "ranking_acc",
                "top1_action_selection",
                "instruction_switch_acc",
                "effect_mae",
                "regret_ece",
            ],
            title="Main Scaling Table",
        ),
        "baseline_comparison": write_table_pair(
            output_dir / "baseline_comparison_table",
            baseline_rows,
            [
                "baseline",
                "run_name",
                "success_rate",
                "ranking_acc",
                "top1_action_selection",
                "instruction_switch_acc",
                "effect_mae",
                "regret_ece",
            ],
            title="Baseline Comparison Table",
        ),
        "ablation": write_table_pair(
            output_dir / "ablation_table",
            ablation_rows,
            [
                "baseline",
                "run_name",
                "success_rate",
                "ranking_acc",
                "top1_action_selection",
                "instruction_switch_acc",
                "effect_mae",
                "regret_ece",
            ],
            title="Ablation Table",
        ),
        "causalstress_per_category": write_table_pair(
            output_dir / "causalstress_per_category_table",
            category_rows,
            [
                "source",
                "category",
                "success",
                "selected_success",
                "failure_rate",
                "selected_failure_rate",
                "instruction_switch",
                "top1",
                "pair_correct",
                "ranking_acc",
            ],
            title="CausalStress Per-Category Table",
        ),
    }

    figure_paths = {
        "performance_vs_k": figures_dir / "performance_vs_k.png",
        "same_state_vs_cross_state_ranking": figures_dir
        / "same_state_vs_cross_state_ranking.png",
        "physical_outcome_vs_label_only": figures_dir / "physical_outcome_vs_label_only.png",
        "success_by_failure_category": figures_dir / "success_by_failure_category.png",
        "regret_calibration": figures_dir / "regret_calibration.png",
    }
    plot_performance_vs_k(scaling_rows, figure_paths["performance_vs_k"])
    plot_same_state_vs_cross_state(baseline_rows, figure_paths["same_state_vs_cross_state_ranking"])
    plot_physical_vs_label_only(baseline_rows, figure_paths["physical_outcome_vs_label_only"])
    plot_success_by_failure_category(category_rows, figure_paths["success_by_failure_category"])
    plot_regret_calibration(metric_rows, scaling_rows, figure_paths["regret_calibration"])

    summary_md = render_result_summary(
        metric_rows=metric_rows,
        scaling_rows=scaling_rows,
        baseline_rows=baseline_rows,
        category_rows=category_rows,
        table_paths=table_paths,
        figure_paths=figure_paths,
    )
    summary_path = output_dir / "result_summary.md"
    summary_path.write_text(summary_md, encoding="utf-8")

    manifest = {
        "runs_dir": str(runs_path),
        "out_dir": str(output_dir),
        "num_payloads": len(payloads),
        "num_metric_rows": len(metric_rows),
        "num_scaling_rows": len(scaling_rows),
        "num_baseline_rows": len(baseline_rows),
        "num_category_rows": len(category_rows),
        "tables": {
            key: {"csv": str(value["csv"]), "markdown": str(value["markdown"])}
            for key, value in table_paths.items()
        },
        "figures": {key: str(path) for key, path in figure_paths.items()},
        "result_summary": str(summary_path),
    }
    write_json(manifest, output_dir / "artifact_manifest.json")
    return manifest


def load_result_payloads(runs_dir: Path) -> list[dict[str, Any]]:
    paths: OrderedDict[str, Path] = OrderedDict()
    for pattern in (
        "**/metrics.json",
        "**/causalstress.json",
        "**/lattice_eval.json",
        "**/scaling_summary.json",
    ):
        for path in sorted(runs_dir.glob(pattern)):
            paths[str(path)] = path
    payloads: list[dict[str, Any]] = []
    for path in paths.values():
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        if isinstance(payload, dict):
            payloads.append({"path": path, "payload": payload})
    return payloads


def collect_metric_rows(payloads: list[dict[str, Any]], runs_dir: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for item in payloads:
        path = Path(item["path"])
        payload = item["payload"]
        if "runs" in payload and isinstance(payload["runs"], dict):
            for key, metrics in payload["runs"].items():
                if isinstance(metrics, dict):
                    rows.append(normalize_metric_row(metrics, path, runs_dir, run_name=f"k{key}"))
            continue
        rows.append(normalize_metric_row(payload, path, runs_dir))
    return sorted(rows, key=lambda row: (_sort_numeric(row.get("k")), row.get("run_name", "")))


def collect_scaling_rows(payloads: list[dict[str, Any]], runs_dir: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = [
        row for row in collect_metric_rows(payloads, runs_dir) if _is_number(row.get("k"))
    ]
    csv_rows = read_scaling_csvs(runs_dir)
    keyed: OrderedDict[str, dict[str, Any]] = OrderedDict()
    for row in rows + csv_rows:
        key = str(row.get("k", row.get("run_name", len(keyed))))
        keyed[key] = {**keyed.get(key, {}), **row}
    return sorted(keyed.values(), key=lambda row: _sort_numeric(row.get("k")))


def collect_baseline_rows(metric_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows = [
        row
        for row in metric_rows
        if row.get("baseline") or row.get("run_name") in BASELINE_NAMES
    ]
    for row in rows:
        if not row.get("baseline") and row.get("run_name") in BASELINE_NAMES:
            row["baseline"] = row["run_name"]
    return sorted(rows, key=lambda row: str(row.get("baseline", row.get("run_name", ""))))


def collect_per_category_rows(payloads: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for item in payloads:
        path = Path(item["path"])
        payload = item["payload"]
        per_category = payload.get("per_category")
        if not isinstance(per_category, dict):
            eval_payload = payload.get("eval")
            if isinstance(eval_payload, dict):
                per_category = eval_payload.get("per_category")
        if not isinstance(per_category, dict):
            continue
        for category, metrics in sorted(per_category.items()):
            if not isinstance(metrics, dict):
                continue
            row = {"source": str(path), "category": category}
            for key, value in metrics.items():
                if _is_number(value):
                    row[key] = float(value)
            if "ranking_acc" not in row and "pair_correct" in row:
                row["ranking_acc"] = row["pair_correct"]
            rows.append(row)
    return rows


def normalize_metric_row(
    payload: dict[str, Any], path: Path, runs_dir: Path, *, run_name: str | None = None
) -> dict[str, Any]:
    config = payload.get("config") if isinstance(payload.get("config"), dict) else {}
    eval_payload = payload.get("eval") if isinstance(payload.get("eval"), dict) else {}
    row: dict[str, Any] = {
        "run_name": run_name
        or str(payload.get("run_name") or payload.get("name") or path.parent.name),
        "source_path": str(path),
        "relative_path": str(_relative_to(path, runs_dir)),
        "baseline": str(payload.get("baseline") or config.get("baseline") or ""),
        "k": _first_present(payload, "k", "K", default=config.get("k", "")),
        "num_states": _first_present(payload, "num_states", default=config.get("num_states", "")),
        "effective_total_records": _first_present(
            payload, "effective_total_records", default=config.get("effective_total_records", "")
        ),
        "checkpoint": str(payload.get("checkpoint") or eval_payload.get("checkpoint") or ""),
    }
    if not row["baseline"]:
        inferred = infer_baseline_from_path(path)
        if inferred:
            row["baseline"] = inferred
    for canonical, aliases in METRIC_ALIASES.items():
        row[canonical] = metric_value(payload, aliases)
    row["score"] = composite_score(row)
    return row


def read_scaling_csvs(runs_dir: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for path in sorted(runs_dir.glob("**/scaling_results.csv")):
        with path.open("r", encoding="utf-8", newline="") as handle:
            for raw in csv.DictReader(handle):
                row = {key: coerce_number(value) for key, value in raw.items()}
                row.setdefault("run_name", f"k{row.get('k', len(rows))}")
                row["source_path"] = str(path)
                row["relative_path"] = str(path)
                rows.append(row)
    return rows


def write_table_pair(
    stem: Path, rows: list[dict[str, Any]], fieldnames: list[str], *, title: str
) -> dict[str, Path]:
    csv_path = stem.with_suffix(".csv")
    md_path = stem.with_suffix(".md")
    write_csv(csv_path, rows, fieldnames)
    md_path.write_text(markdown_table(rows, fieldnames, title=title), encoding="utf-8")
    return {"csv": csv_path, "markdown": md_path}


def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None:
    ensure_dir(path.parent)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            writer.writerow({field: format_cell(row.get(field, "")) for field in fieldnames})


def markdown_table(rows: list[dict[str, Any]], fieldnames: list[str], *, title: str) -> str:
    lines = [f"# {title}", ""]
    lines.append("| " + " | ".join(fieldnames) + " |")
    lines.append("| " + " | ".join("---" for _field in fieldnames) + " |")
    if not rows:
        lines.append("| " + " | ".join("_n/a_" for _field in fieldnames) + " |")
    for row in rows:
        cells = " | ".join(format_cell(row.get(field, "")) for field in fieldnames)
        lines.append(f"| {cells} |")
    return "\n".join(lines).rstrip() + "\n"


def plot_performance_vs_k(rows: list[dict[str, Any]], path: Path) -> None:
    data = [row for row in rows if _is_number(row.get("k"))]
    if not data:
        write_placeholder_png(path)
        return
    plt = pyplot()
    if plt is None:
        write_placeholder_png(path)
        return
    fig, ax = plt.subplots(figsize=(6, 4))
    x_values = [float(row["k"]) for row in data]
    for metric in ("success_rate", "ranking_acc", "instruction_switch_acc"):
        y_values = [float(row.get(metric, 0.0) or 0.0) for row in data]
        ax.plot(x_values, y_values, marker="o", label=metric)
    if all(value > 0 for value in x_values):
        ax.set_xscale("log", base=2)
    ax.set_xlabel("K")
    ax.set_ylabel("Score")
    ax.set_title("Performance vs K")
    ax.grid(True, alpha=0.3)
    ax.legend()
    save_figure(fig, plt, path)


def plot_same_state_vs_cross_state(rows: list[dict[str, Any]], path: Path) -> None:
    cross = best_row([row for row in rows if row.get("baseline") == "cross_state_negatives"])
    same = best_row(
        [
            row
            for row in rows
            if row.get("baseline") not in {"cross_state_negatives", "label_only_counterfactual"}
        ]
    )
    labels = ["same_state", "cross_state"]
    values = [
        float((same or {}).get("ranking_acc", 0.0) or 0.0),
        float((cross or {}).get("ranking_acc", 0.0) or 0.0),
    ]
    plot_bar(labels, values, path, title="Same-State vs Cross-State Ranking", ylabel="ranking_acc")


def plot_physical_vs_label_only(rows: list[dict[str, Any]], path: Path) -> None:
    label = best_row([row for row in rows if row.get("baseline") == "label_only_counterfactual"])
    physical = best_row([row for row in rows if row.get("baseline") != "label_only_counterfactual"])
    labels = ["physical_outcome", "label_only"]
    values = [
        float((physical or {}).get("ranking_acc", 0.0) or 0.0),
        float((label or {}).get("ranking_acc", 0.0) or 0.0),
    ]
    plot_bar(labels, values, path, title="Physical Outcome vs Label-Only", ylabel="ranking_acc")


def plot_success_by_failure_category(rows: list[dict[str, Any]], path: Path) -> None:
    labels = [str(row.get("category", "")) for row in rows]
    values = [
        float(
            row.get(
                "selected_success",
                row.get("success", 1.0 - float(row.get("failure_rate", 1.0))),
            )
            or 0.0
        )
        for row in rows
    ]
    plot_bar(labels, values, path, title="Success by Failure Category", ylabel="success")


def plot_regret_calibration(
    metric_rows: list[dict[str, Any]], scaling_rows: list[dict[str, Any]], path: Path
) -> None:
    rows = scaling_rows or metric_rows
    if not rows:
        write_placeholder_png(path)
        return
    if any(_is_number(row.get("k")) for row in rows):
        plot_line(
            [float(row.get("k") or index + 1) for index, row in enumerate(rows)],
            [float(row.get("regret_ece", 0.0) or 0.0) for row in rows],
            path,
            title="Regret Calibration",
            xlabel="K",
            ylabel="regret_ece",
            log_x=True,
        )
    else:
        plot_bar(
            [str(row.get("run_name", index + 1)) for index, row in enumerate(rows)],
            [float(row.get("regret_ece", 0.0) or 0.0) for row in rows],
            path,
            title="Regret Calibration",
            ylabel="regret_ece",
        )


def plot_bar(
    labels: list[str], values: list[float], path: Path, *, title: str, ylabel: str
) -> None:
    plt = pyplot()
    if plt is None or not labels:
        write_placeholder_png(path)
        return
    fig, ax = plt.subplots(figsize=(max(6, min(12, len(labels) * 0.8)), 4))
    ax.bar(labels, values)
    ax.set_title(title)
    ax.set_ylabel(ylabel)
    ax.tick_params(axis="x", rotation=30)
    fig.tight_layout()
    save_figure(fig, plt, path)


def plot_line(
    x_values: list[float],
    y_values: list[float],
    path: Path,
    *,
    title: str,
    xlabel: str,
    ylabel: str,
    log_x: bool = False,
) -> None:
    plt = pyplot()
    if plt is None or not x_values:
        write_placeholder_png(path)
        return
    fig, ax = plt.subplots(figsize=(6, 4))
    ax.plot(x_values, y_values, marker="o")
    if log_x and all(value > 0 for value in x_values):
        ax.set_xscale("log", base=2)
    ax.set_title(title)
    ax.set_xlabel(xlabel)
    ax.set_ylabel(ylabel)
    ax.grid(True, alpha=0.3)
    save_figure(fig, plt, path)


def render_result_summary(
    *,
    metric_rows: list[dict[str, Any]],
    scaling_rows: list[dict[str, Any]],
    baseline_rows: list[dict[str, Any]],
    category_rows: list[dict[str, Any]],
    table_paths: dict[str, dict[str, Path]],
    figure_paths: dict[str, Path],
) -> str:
    best = best_row(metric_rows)
    warnings = expected_claim_warnings(scaling_rows, baseline_rows)
    lines = [
        "# DoVLA-CIL Paper Artifact Summary",
        "",
        "## Key Claims",
        "",
    ]
    if best:
        lines.append(
            "- Best detected model: "
            f"`{best.get('run_name')}` with ranking_acc={format_cell(best.get('ranking_acc'))} "
            f"and success_rate={format_cell(best.get('success_rate'))}."
        )
    else:
        lines.append("- No model metrics were found.")
    lines.extend(
        [
            f"- Scaling rows collected: {len(scaling_rows)}.",
            f"- Baseline rows collected: {len(baseline_rows)}.",
            f"- CausalStress category rows collected: {len(category_rows)}.",
            "",
            "## Tables",
            "",
        ]
    )
    for key, paths in sorted(table_paths.items()):
        lines.append(f"- `{key}`: `{paths['csv']}`, `{paths['markdown']}`")
    lines.extend(["", "## Figures", ""])
    for key, path in sorted(figure_paths.items()):
        lines.append(f"- `{key}`: `{path}`")
    lines.extend(["", "## Expected Claim Checks", ""])
    if warnings:
        for warning in warnings:
            lines.append(f"- Warning: {warning}")
    else:
        lines.append("- No automatic warnings for the expected claims.")
    return "\n".join(lines).rstrip() + "\n"


def expected_claim_warnings(
    scaling_rows: list[dict[str, Any]], baseline_rows: list[dict[str, Any]]
) -> list[str]:
    warnings: list[str] = []
    if len(scaling_rows) >= 2:
        ordered = sorted(scaling_rows, key=lambda row: _sort_numeric(row.get("k")))
        first = float(ordered[0].get("ranking_acc", 0.0) or 0.0)
        last = float(ordered[-1].get("ranking_acc", 0.0) or 0.0)
        if last <= first:
            warnings.append("ranking accuracy does not improve from smallest K to largest K.")
    else:
        warnings.append("scaling claim is unsupported because fewer than two K values were found.")

    cross = best_row(
        [row for row in baseline_rows if row.get("baseline") == "cross_state_negatives"]
    )
    same = best_row(
        [
            row
            for row in baseline_rows
            if row.get("baseline") not in {"cross_state_negatives", "label_only_counterfactual"}
        ]
    )
    if cross and same:
        same_rank = float(same.get("ranking_acc", 0.0) or 0.0)
        cross_rank = float(cross.get("ranking_acc", 0.0) or 0.0)
        if same_rank <= cross_rank:
            warnings.append("same-state ranking does not beat cross-state negatives.")
    else:
        warnings.append("same-state vs cross-state comparison is missing a required baseline.")

    label = best_row(
        [row for row in baseline_rows if row.get("baseline") == "label_only_counterfactual"]
    )
    physical = best_row(
        [row for row in baseline_rows if row.get("baseline") != "label_only_counterfactual"]
    )
    if label and physical:
        physical_rank = float(physical.get("ranking_acc", 0.0) or 0.0)
        label_rank = float(label.get("ranking_acc", 0.0) or 0.0)
        if physical_rank <= label_rank:
            warnings.append("physical-outcome baseline does not beat label-only counterfactuals.")
    else:
        warnings.append("physical outcome vs label-only comparison is missing a required baseline.")
    return warnings


def metric_value(payload: dict[str, Any], aliases: Iterable[str]) -> float:
    search_payloads = [payload]
    if isinstance(payload.get("eval"), dict):
        search_payloads.append(payload["eval"])
    if isinstance(payload.get("metrics"), dict):
        search_payloads.append(payload["metrics"])
    for item in search_payloads:
        for alias in aliases:
            value = lookup_dotted(item, alias)
            if _is_number(value):
                return float(value)
    return 0.0


def lookup_dotted(payload: dict[str, Any], key: str) -> Any:
    if key in payload:
        return payload[key]
    cursor: Any = payload
    for part in key.split("."):
        if not isinstance(cursor, dict) or part not in cursor:
            return None
        cursor = cursor[part]
    return cursor


def _first_present(payload: dict[str, Any], *keys: str, default: Any = "") -> Any:
    for key in keys:
        if key in payload and payload[key] not in (None, ""):
            return payload[key]
    return default if default is not None else ""


def infer_baseline_from_path(path: Path) -> str:
    for part in path.parts:
        if part in BASELINE_NAMES:
            return part
    return ""


def composite_score(row: dict[str, Any]) -> float:
    values = [
        float(row.get("success_rate", 0.0) or 0.0),
        float(row.get("ranking_acc", 0.0) or 0.0),
        float(row.get("top1_action_selection", 0.0) or 0.0),
        float(row.get("instruction_switch_acc", 0.0) or 0.0),
    ]
    return sum(values) / len(values)


def best_row(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
    if not rows:
        return None
    return max(
        rows,
        key=lambda row: (
            float(row.get("ranking_acc", 0.0) or 0.0),
            float(row.get("success_rate", 0.0) or 0.0),
            float(row.get("score", 0.0) or 0.0),
        ),
    )


def coerce_number(value: Any) -> Any:
    if isinstance(value, int | float):
        return value
    if isinstance(value, str):
        stripped = value.strip()
        if not stripped:
            return ""
        try:
            if any(char in stripped for char in ".eE"):
                return float(stripped)
            return int(stripped)
        except ValueError:
            return value
    return value


def format_cell(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, float):
        if math.isnan(value) or math.isinf(value):
            return ""
        return f"{value:.6g}"
    return str(value)


def _is_number(value: Any) -> bool:
    return isinstance(value, int | float) and not isinstance(value, bool)


def _sort_numeric(value: Any) -> tuple[int, float, str]:
    if _is_number(value):
        return (0, float(value), "")
    try:
        return (0, float(str(value)), "")
    except (TypeError, ValueError):
        return (1, 0.0, str(value))


def _relative_to(path: Path, parent: Path) -> Path:
    try:
        return path.relative_to(parent)
    except ValueError:
        return path


def pyplot():
    try:
        import matplotlib

        matplotlib.use("Agg")
        import matplotlib.pyplot as plt
    except ImportError:
        return None
    return plt


def save_figure(fig: Any, plt: Any, path: Path) -> None:
    ensure_dir(path.parent)
    fig.tight_layout()
    fig.savefig(path)
    plt.close(fig)


def write_placeholder_png(path: Path) -> None:
    ensure_dir(path.parent)
    width, height = 1, 1
    raw = b"\x00\xff\xff\xff"
    png = b"\x89PNG\r\n\x1a\n"
    png += png_chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
    png += png_chunk(b"IDAT", zlib.compress(raw))
    png += png_chunk(b"IEND", b"")
    path.write_bytes(png)


def png_chunk(kind: bytes, data: bytes) -> bytes:
    return (
        struct.pack(">I", len(data))
        + kind
        + data
        + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
    )


if __name__ == "__main__":
    raise SystemExit(main())