sophia/bin/dev

#!/usr/bin/env python3

import json
import os
import re
import shlex
import shutil
import struct
import subprocess
import sys

COMMAND_TO_METHOD = {
    "branch-from": "branch_from",
    "bash-completions": "bash_completions",
}
METHOD_TO_COMMAND = {method: command for command, method in COMMAND_TO_METHOD.items()}

WAIT_FOR_PROCESS = "wait_for_process"

class DevHelper:
    SHADER_EXTENSIONS = (".vert", ".frag", ".comp", ".geom", ".tesc", ".tese")
    PROJECT_NAME = "sophia"
    DEFAULT_PRESET = "debug"

    def __init__(self):
        self.last_exit_status = 0

    # Core logic
    def call(self, *args, **kwargs):
        if not args:
            print("Usage: dev <command> [args...]\n\nCommands:")
            for command in self._commands():
                print(f"  {command}")
            return

        command, rest = args[0], args[1:]
        method_name = COMMAND_TO_METHOD.get(command, command)
        method = getattr(self, method_name, None)
        if callable(method) and not method_name.startswith("_") and method_name != "call":
            return method(*rest, **kwargs)

    # Primary command wrappers

    ##
    # Configures and builds a CMake preset (see CMakePresets.json).
    #   dev build [preset] [--fresh]
    # The preset defaults to debug and builds into build/<preset>. --fresh
    # discards the preset's CMake cache before configuring.
    def build(self, *args, **kwargs):
        preset, fresh = self._parse_preset_args(args, "build")
        self._build_preset(preset, fresh)

    ##
    # Builds the debug preset (incrementally), then runs the editor from
    # build/debug. Any arguments are passed through to the editor.
    #   dev editor [args...]
    def editor(self, *args, **kwargs):
        preset = self.DEFAULT_PRESET
        self._build_preset(preset, fresh=False)

        preset_dir = os.path.join(self._project_root(), "build", preset)
        binary = os.path.join(preset_dir, "bin", "editor")
        print(f"Running: {shlex.join([binary, *args])}")
        os.chdir(preset_dir)
        os.execv(binary, [binary, *args])

    ##
    # Builds the debug preset (incrementally), then runs the testbed from
    # build/debug. Any arguments are passed through to the testbed.
    #   dev testbed [args...]
    def testbed(self, *args, **kwargs):
        preset = self.DEFAULT_PRESET
        self._build_preset(preset, fresh=False)

        preset_dir = os.path.join(self._project_root(), "build", preset)
        binary = os.path.join(preset_dir, "bin", "testbed")
        print(f"Running: {shlex.join([binary, *args])}")
        os.chdir(preset_dir)
        os.execv(binary, [binary, *args])

    ##
    # Checks out a branch named after a Gitea issue URL.
    # If the issue is from this project's repo, the branch name will be:
    #   issue-<issue_number>/<issue_title>
    # If the issue is from another repo (presumably upstream), the branch name will be:
    #   <issue_owner>-issue-<issue_number>/<issue_title>
    def branch_from(self, *args, **kwargs):
        issue_url = args[0] if args else None
        if issue_url is None:
            raise ValueError("Must provide a Gitea issue URL.")

        base_url, issue_repo, issue_number = self._parse_issue_url(issue_url)
        issue_title = self._fetch_issue_title(base_url, issue_repo, issue_number)
        branch_name = self._format_branch_name(issue_repo, issue_number, issue_title)

        exists = subprocess.run(
            ["git", "-C", self._project_root(), "rev-parse", "--verify", "--quiet", f"refs/heads/{branch_name}"],
            stdout=subprocess.DEVNULL,
        ).returncode == 0
        flag = "" if exists else "-b "
        self._exec_in_root(f"git checkout {flag}{shlex.quote(branch_name)}", replace=True)

    ##
    # Creates an issue on this project's Gitea repo.
    #   dev issue <title> [-b|--body <text>] [-l|--label <name>]... [-a|--assignee <user>]... [-o|--open]
    # The body is read from stdin if --body is not given and stdin is piped.
    # Gitea's API cannot add issues to a project (only the web UI can), so the
    # issue URL is printed and --open opens it in the browser to do so.
    def issue(self, *args, **kwargs):
        title, body, labels, assignees, open_in_browser = self._parse_issue_args(args)

        repo = self._origin_repo()
        if repo is None:
            raise RuntimeError("could not determine the Gitea repo from the origin remote")

        payload = {"title": title, "body": body}
        if labels:
            label_ids = self._fetch_label_ids(repo)
            unknown = [label for label in labels if label not in label_ids]
            if unknown:
                raise ValueError(f"Unknown label(s): {', '.join(unknown)} (available: {', '.join(sorted(label_ids))})")
            payload["labels"] = [label_ids[label] for label in labels]
        if assignees:
            payload["assignees"] = assignees

        issue = self._tea_api(f"repos/{repo}/issues", payload)
        print(f"Created issue #{issue['number']}: {issue['title']}")
        print(issue["html_url"])
        print(f"Add it to the project from the issue sidebar: Projects > {self.PROJECT_NAME}")

        if open_in_browser:
            opener = shutil.which("xdg-open") or shutil.which("open")
            if opener is None:
                raise RuntimeError("xdg-open not found on PATH")
            subprocess.Popen([opener, issue["html_url"]], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    def renderdoc(self, *args, **kwargs):
        qrenderdoc = self._find_qrenderdoc()
        self._exec_in_root(qrenderdoc, replace=True)

    ##
    # Builds the Tracy profiler (viewer) from the third_party/tracy submodule,
    # then runs it. Building from the submodule keeps the viewer's version in
    # sync with the client compiled into the engine.
    #   dev tracy [--fresh] [args...]
    # Configures only when there is no build yet (or --fresh), then builds
    # incrementally. Other arguments are passed through to the profiler.
    def tracy(self, *args, **kwargs):
        fresh = "--fresh" in args
        args = [arg for arg in args if arg != "--fresh"]

        source_dir = os.path.join("third_party", "tracy", "profiler")
        build_dir = os.path.join(source_dir, "build")
        if fresh or not os.path.isfile(os.path.join(self._project_root(), build_dir, "CMakeCache.txt")):
            configure = ["cmake", "-B", build_dir, "-S", source_dir, "-DCMAKE_BUILD_TYPE=Release"]
            if fresh:
                configure.append("--fresh")
            self._run_in_root(configure)
        self._run_in_root(["cmake", "--build", build_dir, "--parallel"])

        binary = os.path.join(self._project_root(), build_dir, "tracy-profiler")
        print(f"Running: {shlex.join([binary, *args])}")
        os.execv(binary, [binary, *args])

    def shaders(self, *args, **kwargs):
        shaders_dir = os.path.join(self._project_root(), "engine", "src", "shaders")
        if not os.path.isdir(shaders_dir):
            raise ValueError(f"Shaders directory not found: {shaders_dir}")

        glslc = self._find_glslc()

        shader_paths = sorted(
            os.path.join(root, filename)
            for root, _dirs, files in os.walk(shaders_dir)
            for filename in files
            if filename.endswith(self.SHADER_EXTENSIONS)
        )

        if not shader_paths:
            print(f"No shaders found in {shaders_dir}")
            return

        for shader_path in shader_paths:
            self._generate_shader_header(glslc, shader_path)

    # Custom helpers

    def ownit(self, *args, **kwargs):
        file_or_directory = args[0] if args else None
        if file_or_directory is None:
            raise ValueError("Must provide a file or directory path.")

        if sys.platform.startswith("linux"):
            print(f"Take ownership of the file or directory? {file_or_directory}")
            command = f"sudo chown -R {self._user_id()}:{self._group_id()} {file_or_directory}"
            self._exec(command, replace=True)
        else:
            raise NotImplementedError(f"Not implemented for platform {sys.platform}")

    def bash_completions(self, *args, **kwargs):
        print(" ".join(self._commands()))

    # Private helpers
    def _commands(self):
        return sorted(
            METHOD_TO_COMMAND.get(name, name)
            for name in vars(DevHelper)
            if not name.startswith("_") and name != "call" and callable(getattr(DevHelper, name))
        )

    def _parse_preset_args(self, args, command):
        preset = None
        fresh = False
        for arg in args:
            if arg == "--fresh":
                fresh = True
            elif arg.startswith("-"):
                raise ValueError(f"Unknown option: {arg}")
            elif preset is None:
                preset = arg
            else:
                raise ValueError(f"Usage: dev {command} [preset] [--fresh]")

        preset = preset or self.DEFAULT_PRESET
        presets = self._configure_presets()
        if preset not in presets:
            raise ValueError(f"Unknown preset: {preset} (available: {', '.join(presets)})")
        return preset, fresh

    def _configure_presets(self):
        with open(os.path.join(self._project_root(), "CMakePresets.json")) as f:
            presets = json.load(f)["configurePresets"]
        return [preset["name"] for preset in presets if not preset.get("hidden")]

    def _build_preset(self, preset, fresh):
        configure = ["cmake", "--preset", preset]
        if fresh:
            configure.append("--fresh")
        self._run_in_root(configure)
        self._copy_compile_commands(preset)
        self._run_in_root(["cmake", "--build", "--preset", preset])

    # Editor tooling reads build/compile_commands.json, so point it at the
    # most recently configured preset.
    def _copy_compile_commands(self, preset):
        build_dir = os.path.join(self._project_root(), "build")
        source = os.path.join(build_dir, preset, "compile_commands.json")
        if os.path.isfile(source):
            shutil.copyfile(source, os.path.join(build_dir, "compile_commands.json"))

    def _run_in_root(self, command):
        print(f"Running: {shlex.join(command)}")
        result = subprocess.run(command, cwd=self._project_root())
        if result.returncode != 0:
            sys.exit(result.returncode)

    def _parse_issue_url(self, url):
        match = re.match(r"^(https?://[^/]+)/([^/]+)/([^/]+)/issues/(\d+)", url)
        if match is None:
            raise ValueError(f"Not a Gitea issue URL: {url}")
        return match[1], f"{match[2]}/{match[3]}", match[4]

    def _fetch_issue_title(self, base_url, repo, issue_number):
        if shutil.which("tea") is None:
            raise RuntimeError("tea not found on PATH (install gitea tea and run `tea login add`)")

        endpoint = f"{base_url}/api/v1/repos/{repo}/issues/{issue_number}"
        print(f"Running: tea api {endpoint}")
        result = subprocess.run(["tea", "api", endpoint], stdout=subprocess.PIPE, text=True)
        if result.returncode != 0:
            raise RuntimeError(f"failed to fetch issue title from {endpoint}")
        return json.loads(result.stdout)["title"]

    def _parse_issue_args(self, args):
        title = None
        body = None
        labels = []
        assignees = []
        open_in_browser = False

        args = list(args)
        while args:
            arg = args.pop(0)
            if arg in ("-b", "--body", "-l", "--label", "-a", "--assignee"):
                if not args:
                    raise ValueError(f"{arg} requires a value")
                value = args.pop(0)
                if arg in ("-b", "--body"):
                    body = value
                elif arg in ("-l", "--label"):
                    labels.append(value)
                else:
                    assignees.append(value)
            elif arg in ("-o", "--open"):
                open_in_browser = True
            elif arg.startswith("-"):
                raise ValueError(f"Unknown option: {arg}")
            elif title is None:
                title = arg
            else:
                title = f"{title} {arg}"

        if not title:
            raise ValueError("Must provide an issue title. Usage: dev issue <title> [-b <body>] [-l <label>]... [-a <user>]... [-o]")
        if body is None:
            body = "" if sys.stdin.isatty() else sys.stdin.read()
        return title, body, labels, assignees, open_in_browser

    def _tea_api(self, endpoint, payload=None):
        if shutil.which("tea") is None:
            raise RuntimeError("tea not found on PATH (install gitea tea and run `tea login add`)")

        command = ["tea", "api", endpoint]
        if payload is not None:
            command += ["-d", "@-"]
        print(f"Running: {' '.join(command)}")
        result = subprocess.run(
            command,
            input=json.dumps(payload) if payload is not None else None,
            stdout=subprocess.PIPE,
            text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(f"tea api request failed: {endpoint}")
        return json.loads(result.stdout)

    def _fetch_label_ids(self, repo):
        labels = self._tea_api(f"repos/{repo}/labels?limit=100")
        return {label["name"]: label["id"] for label in labels}

    def _format_branch_name(self, issue_repo, issue_number, issue_title):
        formatted_title = re.sub(r"-+", "-", re.sub(r"[^a-z0-9\-]", "", re.sub(r"\s+", "-", issue_title.lower().strip())))
        if issue_repo == self._origin_repo():
            return f"issue-{issue_number}/{formatted_title}"
        issue_owner = issue_repo.split("/")[0]
        return f"{issue_owner}-issue-{issue_number}/{formatted_title}"

    def _origin_repo(self):
        result = subprocess.run(
            ["git", "-C", self._project_root(), "remote", "get-url", "origin"],
            stdout=subprocess.PIPE,
            text=True,
        )
        match = re.search(r"[:/]([^/:]+/[^/]+?)(?:\.git)?$", result.stdout.strip())
        return match[1] if match else None

    def _find_glslc(self):
        glslc = shutil.which("glslc")
        if glslc is None:
            raise RuntimeError(
                "glslc not found on PATH (install the Vulkan SDK)"
            )
        return glslc

    def _find_qrenderdoc(self):
        qrenderdoc = shutil.which("qrenderdoc")
        if qrenderdoc is None:
            raise RuntimeError(
                "qrenderdoc not found on PATH (install RenderDoc)"
            )
        return qrenderdoc

    def _generate_shader_header(self, glslc, shader_path):
        relative_path = os.path.relpath(shader_path, self._project_root())
        print(f"Compiling shader: {relative_path}")

        result = subprocess.run(
            [glslc, shader_path, "-o", "-"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"failed to compile {shader_path}: {result.stderr.decode()}"
            )

        spirv = result.stdout
        if len(spirv) % 4 != 0:
            raise RuntimeError(
                f"invalid SPIR-V output for {shader_path} (size not a multiple of 4)"
            )

        word_count = len(spirv) // 4
        words = struct.unpack(f"<{word_count}I", spirv)

        filename = os.path.basename(shader_path)
        variable_name = f"{filename.replace('.', '_')}_spv"
        header_path = shader_path + ".spv.h"

        body_lines = [
            "    " + ", ".join(f"0x{word:08x}" for word in words[i:i + 4]) + ","
            for i in range(0, len(words), 4)
        ]

        header = "\n".join([
            "#pragma once",
            "",
            "#include <array>",
            "#include <cstdint>",
            "",
            "namespace sophia::shaders {",
            "",
            f"inline constexpr std::array<uint32_t, {word_count}> {variable_name} = {{",
            *body_lines,
            "};",
            "",
            "}  // namespace sophia::shaders",
            "",
        ])

        with open(header_path, "w") as f:
            f.write(header)

        print(f"Wrote {os.path.relpath(header_path, self._project_root())} "
              f"({word_count * 4} bytes)")

    def _wait_for_process_with_logging(self, command):
        process = self._exec(command, capture=True)
        for line in process.stdout:
            print(line, end="")
        process.wait()
        self.last_exit_status = process.returncode

    def _exec_in_root(self, command, **kwargs):
        return self._exec(f"cd {self._project_root()} && {command}", **kwargs)

    def _exec(self, command, *, replace=False, capture=False):
        print(f"Running: {command}")
        if replace:
            os.execv("/bin/sh", ["/bin/sh", "-c", command])
        if capture:
            return subprocess.Popen(
                ["/bin/sh", "-c", f"{command} 2>&1"],
                stdout=subprocess.PIPE,
                text=True,
            )
        return subprocess.run(["/bin/sh", "-c", command])

    def _project_root(self):
        return os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))

    def _take_over_needed(self, file_or_directory):
        result = subprocess.run(
            ["/bin/sh", "-c", f"find {file_or_directory} -not -user {self._user_id()} -print -quit | grep -q ."],
        )
        return result.returncode == 0

    def _user_id(self):
        if not sys.platform.startswith("linux"):
            raise NotImplementedError(f"Not implemented for platform {sys.platform}")
        return subprocess.run(["id", "-u"], stdout=subprocess.PIPE, text=True).stdout.strip()

    def _group_id(self):
        if not sys.platform.startswith("linux"):
            raise NotImplementedError(f"Not implemented for platform {sys.platform}")
        return subprocess.run(["id", "-g"], stdout=subprocess.PIPE, text=True).stdout.strip()


if __name__ == "__main__":
    DevHelper().call(*sys.argv[1:])