Blacksite/scripts/editor/build_progress.py

#!/usr/bin/env python3
"""Build the Blacksite editor while publishing exact crate progress."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import subprocess
from typing import Iterable


def write_status(
    path: Path,
    phase: str,
    detail: str,
    progress: int,
    state: str = "running",
    crate_count: str = "",
) -> None:
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(
        f"{phase}\n{detail}\n{progress}\n{state}\n{crate_count}\n",
        encoding="utf-8",
    )
    os.replace(temporary, path)


def unique_crates(tree_output: str) -> set[str]:
    return {line.strip() for line in tree_output.splitlines() if line.strip()}


def completed_crate(
    message: object, completed_package_ids: set[str]
) -> tuple[bool, str]:
    if not isinstance(message, dict) or message.get("reason") != "compiler-artifact":
        return False, ""
    package_id = message.get("package_id")
    if not isinstance(package_id, str) or package_id in completed_package_ids:
        return False, ""
    completed_package_ids.add(package_id)
    target = message.get("target")
    if isinstance(target, dict) and isinstance(target.get("name"), str):
        return True, target["name"]
    return True, package_id.rsplit("#", 1)[-1]


def cargo_tree(cargo: str) -> list[str]:
    return [
        cargo,
        "tree",
        "-p",
        "editor",
        "--features",
        "dev",
        "--edges",
        "normal,build",
        "--prefix",
        "none",
        "--format",
        "{p}",
        "--no-dedupe",
        "--color",
        "never",
    ]


def cargo_build(cargo: str) -> list[str]:
    return [
        cargo,
        "build",
        "-p",
        "editor",
        "--bins",
        "--features",
        "dev",
        "--message-format=json-render-diagnostics",
        "--color",
        "never",
    ]


def rendered_lines(message: object, raw_line: str) -> Iterable[str]:
    if not isinstance(message, dict):
        yield raw_line
        return
    if message.get("reason") == "compiler-message":
        diagnostic = message.get("message")
        if isinstance(diagnostic, dict):
            rendered = diagnostic.get("rendered")
            if isinstance(rendered, str) and rendered:
                yield rendered.rstrip("\n")


def run(status_path: Path, log_path: Path, cargo: str) -> int:
    write_status(
        status_path,
        "Preparing build",
        "Resolving editor crate graph",
        8,
    )
    tree = subprocess.run(
        cargo_tree(cargo),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    crates = unique_crates(tree.stdout) if tree.returncode == 0 else set()
    if not crates:
        with log_path.open("a", encoding="utf-8") as log:
            log.write(tree.stdout)
            log.write(tree.stderr)
        write_status(
            status_path,
            "Build graph failed",
            "Open the launch log for Cargo details",
            8,
            "error",
        )
        return tree.returncode or 1

    total = len(crates)
    write_status(
        status_path,
        "Compiling project",
        "Preparing editor crates",
        12,
        crate_count=f"0/{total}",
    )

    process = subprocess.Popen(
        cargo_build(cargo),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
    )
    assert process.stdout is not None
    completed_package_ids: set[str] = set()
    with log_path.open("a", encoding="utf-8") as log:
        for raw_line in process.stdout:
            stripped = raw_line.rstrip("\n")
            try:
                message: object = json.loads(stripped)
            except json.JSONDecodeError:
                message = stripped

            for rendered in rendered_lines(message, stripped):
                log.write(rendered + "\n")
                log.flush()

            changed, crate_name = completed_crate(message, completed_package_ids)
            if not changed:
                continue
            completed = min(len(completed_package_ids), total)
            progress = min(78, 12 + (66 * completed // total))
            write_status(
                status_path,
                "Compiling project",
                f"{crate_name} ready",
                progress,
                crate_count=f"{completed}/{total}",
            )

    return_code = process.wait()
    completed = min(len(completed_package_ids), total)
    if return_code != 0:
        write_status(
            status_path,
            "Build failed",
            "Open the launch log for compiler details",
            max(12, min(78, 12 + (66 * completed // total))),
            "error",
            f"{completed}/{total}",
        )
        return return_code

    write_status(
        status_path,
        "Build complete",
        "Preparing engine process",
        78,
        crate_count=f"{total}/{total}",
    )
    return 0


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("status_path", type=Path)
    parser.add_argument("log_path", type=Path)
    parser.add_argument("--cargo", default="/usr/bin/cargo")
    arguments = parser.parse_args()
    return run(arguments.status_path, arguments.log_path, arguments.cargo)


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