product-vamp/modulus.py

#!/usr/bin/env python3
# Canonical consumer bootstrap. Copy this file unchanged to a project's modulus.py.
import json
import os
import pathlib
import shlex
import subprocess
import sys

PACKAGE_REF = "modulus/latest"
CONAN_VERSION = "2.26.2"
FETCH_IMAGE = "modulus-fetch:dev"
COMMANDS = (
  "configure",
  "build",
  "test",
  "package",
  "lsp",
  "clangd",
  "format",
  "run",
)
COMMON_DIR = pathlib.Path("build") / "build_common"
TOOLS_DIR = pathlib.Path("build") / "build_tools"
RUNTIME_DIR = COMMON_DIR / "modulus"
CONAN_HOME_DIR = COMMON_DIR / "conan_cache"
DEFAULT_REMOTE_NAME = "khodo"
DEFAULT_REMOTE_URL = "https://git.thyme-labs.com/api/packages/khodo/conan"
DEFAULT_DOCKER_MARKER = "/opt/modulus-build-image"

def repo_root():
  root = pathlib.Path(__file__).resolve().parent
  if not (root / "modulus.json").is_file():
    raise SystemExit("modulus.py must be run from a Modulus-managed repository root")
  return root

def read_json(path):
  try:
    data = json.loads(path.read_text(encoding="utf-8"))
  except (OSError, json.JSONDecodeError) as error:
    raise SystemExit(f"failed to read {path}: {error}") from error
  if not isinstance(data, dict):
    raise SystemExit(f"{path} must contain a JSON object")
  return data

def remote_config(root):
  config = read_json(root / "modulus.json")
  remote = config.get(
    "tooling",
    {},
  ).get("remote", {})
  if not isinstance(remote, dict):
    remote = {}
  return (
    str(remote.get("name") or DEFAULT_REMOTE_NAME),
    str(remote.get("url") or DEFAULT_REMOTE_URL),
  )

def runtime_ready(root):
  return (
    (root / TOOLS_DIR / "configure.py").is_file()
    and (root / TOOLS_DIR / "lsp.py").is_file()
    and (root / TOOLS_DIR / "clangd.py").is_file()
    and (root / RUNTIME_DIR / "modulus" / "cli.py").is_file()
    and (root / RUNTIME_DIR / "modulus" / "export.py").is_file()
    and (root / TOOLS_DIR / "format.py").is_file()
    and (root / RUNTIME_DIR / "modulus" / "formatting.py").is_file()
    and (root / RUNTIME_DIR / "modulus" / "format_layout.py").is_file()
  )

def env_flag(name):
  return os.environ.get(name, "").lower() in {"1", "true", "yes", "on"}

def docker_marker(root):
  config = read_json(root / "modulus.json")
  marker = config.get(
    "tooling",
    {},
  ).get("docker", {}).get("marker", DEFAULT_DOCKER_MARKER)
  return pathlib.Path(str(marker))

def direct_mode(root):
  return env_flag("MODULUS_DIRECT") or docker_marker(root).exists()

def run(command, input_text=None, capture=False, display_command=None, env=None, check=True):
  visible = display_command if display_command is not None else command
  print("+", " ".join(shlex.quote(str(part)) for part in visible))
  completed = subprocess.run(
    [str(part) for part in command],
    input=input_text,
    stdout=subprocess.PIPE if capture else None,
    stderr=subprocess.PIPE if capture else None,
    text=True,
    check=False,
    env=env,
  )
  if check and completed.returncode != 0:
    if capture and completed.stdout:
      print(completed.stdout, end="")
    if capture and completed.stderr:
      print(completed.stderr, end="", file=sys.stderr)
    raise SystemExit(completed.returncode)
  return completed

def fetch_dockerfile():
  return f"""FROM debian:bookworm-slim

ENV DEBIAN_FRONTEND=noninteractive

RUN apt-get update \\
  && apt-get install -y --no-install-recommends \\
    ca-certificates \\
    python3 \\
    python3-venv \\
  && rm -rf /var/lib/apt/lists/*

RUN python3 -m venv /opt/modulus-python

ENV PATH=/opt/modulus-python/bin:$PATH

RUN python -m pip install --no-cache-dir conan=={CONAN_VERSION}

WORKDIR /workspace
"""

def ensure_fetch_image(offline=False):
  inspected = subprocess.run(
    [
      "docker",
      "image",
      "inspect",
      FETCH_IMAGE,
    ],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
    check=False,
  )
  if inspected.returncode == 0 and (offline or not docker_platform_args()):
    return
  if offline:
    raise SystemExit('offline bootstrap requires the existing Modulus fetch image or MODULUS_DIRECT=1 with Conan installed')
  run(
    [
      "docker",
      "build",
      *docker_platform_args(),
      "--tag",
      FETCH_IMAGE,
      "-",
    ],
    input_text=fetch_dockerfile(),
  )

def in_container():
  return pathlib.Path("/.dockerenv").exists()

def docker_platform_args():
  platform = os.environ.get("MODULUS_DOCKER_PLATFORM", "").strip()
  return ["--platform", platform] if platform else []

def docker_workspace_args(root):
  if in_container():
    container = os.environ.get("HOSTNAME", "").strip()
    if not container:
      raise SystemExit("HOSTNAME is required to share the CI workspace with the Modulus fetch container")
    return [
      "--volumes-from",
      container,
    ]
  return [
    "--volume",
    f"{root}:{root}",
  ]

def fetch_env(root):
  env = os.environ.copy()
  for name in list(env):
    if name in {"CONAN_USER", "CONAN_TOKEN"} or name.startswith(("CONAN_LOGIN_USERNAME", "CONAN_PASSWORD")):
      del env[name]
  env["CONAN_HOME"] = str(root / CONAN_HOME_DIR)
  env["CONAN_NON_INTERACTIVE"] = "1"
  return env

def run_fetch(root, command, capture=False, display_command=None, check=True, offline=False):
  if direct_mode(root):
    return run(command, capture=capture, display_command=display_command, env=fetch_env(root), check=check)
  ensure_fetch_image(offline=offline)
  docker_command = [
    "docker",
    "run",
    "--rm",
    "--interactive",
    *(["--pull=never", "--network=none"] if offline else []),
    *docker_platform_args(),
    "--user",
    f"{os.getuid()}:{os.getgid()}",
    "--env",
    "HOME=/tmp",
    "--env",
    "CONAN_HOME",
    "--env",
    "CONAN_NON_INTERACTIVE",
    *docker_workspace_args(root),
    "--workdir",
    str(root),
    FETCH_IMAGE,
    *command,
  ]
  return run(docker_command, capture=capture, display_command=display_command, env=fetch_env(root), check=check)

def configure_remote(root, remote_name, remote_url):
  run_fetch(
    root,
    [
      "conan",
      "remote",
      "add",
      "-f",
      remote_name,
      remote_url,
    ],
  )
  run_fetch(root, ["conan", "remote", "logout", remote_name])

def package_refs(data):
  refs = []
  for location_data in data.values():
    if not isinstance(location_data, dict):
      continue
    for ref, ref_data in location_data.items():
      if not ref.startswith(PACKAGE_REF) or not isinstance(ref_data, dict):
        continue
      revisions = ref_data.get(
        "revisions",
        {},
      )
      if not isinstance(revisions, dict):
        continue
      for recipe_revision, revision_data in revisions.items():
        if not isinstance(revision_data, dict):
          continue
        packages = revision_data.get(
          "packages",
          {},
        )
        if not isinstance(packages, dict):
          continue
        for package_id, package_data in packages.items():
          package_revisions = package_data.get(
            "revisions",
            {},
          )if isinstance(package_data, dict) else {}
          if isinstance(package_revisions, dict) and package_revisions:
            for package_revision in package_revisions:
              refs.append(f"{PACKAGE_REF}#{recipe_revision}:{package_id}#{package_revision}")
          else:
            refs.append(f"{PACKAGE_REF}#{recipe_revision}:{package_id}")
  return sorted(refs)

def find_modulus(root, remote_name=None):
  completed = run_fetch(
    root,
    [
      "conan",
      "list",
      PACKAGE_REF + "#latest:*#latest",
      *(["-r", remote_name] if remote_name else ["-c"]),
      "-f",
      "json",
    ],
    capture=True,
    offline=remote_name is None,
  )
  refs = package_refs(json.loads(completed.stdout))
  if not refs:
    raise SystemExit(f"no {PACKAGE_REF} package available in {remote_name or 'the local Conan cache'}")
  package_ref = refs[-1]
  if remote_name:
    run_fetch(root, ["conan", "download", package_ref, "-r", remote_name])
  completed = run_fetch(
    root,
    [
      "conan",
      "cache",
      "path",
      package_ref,
    ],
    capture=True,
    offline=remote_name is None,
  )
  package_root = pathlib.Path(completed.stdout.strip())
  if not (package_root / "modulus" / "export.py").is_file() or not (package_root / "tools").is_dir():
    raise SystemExit(f"downloaded {PACKAGE_REF} package does not contain Modulus export files: {package_root}")
  return package_root

def materialize_runtime(root, no_remote=False):
  remote_name, remote_url = remote_config(root)
  if not no_remote:
    configure_remote(root, remote_name, remote_url)
  package_root = find_modulus(root, None if no_remote else remote_name)
  sys.path.insert(0, str(package_root))
  import modulus.export
  modulus.export.sync_project(root, package_root)

def ensure_runtime(root, update=False, no_remote=False):
  if update or not runtime_ready(root):
    materialize_runtime(root, no_remote=no_remote)

def delegate(command, argv):
  root = repo_root()
  if command in {"clangd", "format"}:
    if not runtime_ready(root):
      print(f"Modulus {command} tooling is missing; run ./modulus.py configure first", file=sys.stderr)
      return 1
  else:
    ensure_runtime(root, update=command == "configure" and "--update" in argv, no_remote="--no-remote" in argv)
  sys.path.insert(0, str(root / RUNTIME_DIR))
  import modulus.cli
  previous_argv = sys.argv
  previous_cwd = pathlib.Path.cwd()
  sys.argv = [
    str(root / "modulus.py"),
    *argv,
  ]
  os.chdir(root)
  try:
    return modulus.cli.main(command)
  finally:
    os.chdir(previous_cwd)
    sys.argv = previous_argv

def main(argv):
  if not argv or argv[0] in ("-h", "--help"):
    print("usage: python3 modulus.py " + "|".join(COMMANDS) + " ...")
    return 0 if argv else 2
  command = argv[0]
  if command not in COMMANDS:
    raise SystemExit(f"unknown Modulus command: {command}")
  return delegate(command, argv[1:])

if __name__ == "__main__":
  raise SystemExit(main(sys.argv[1:]))