# Released under the MIT License. See LICENSE for details.
#
"""Generate C++ id-enum + load-block code for the construct asset-package.
Fetches the asset *listing* for the projectconfig ``"assets"`` pin via
``tools/bacloud assetpackage _listing`` -- an assembly-free server query
that returns every logical asset + kind from the package's workspace
snapshot (no recipes run, no blobs assembled). The same server-side
enumeration backs Python wrapper generation, so the C++ id enums and the
wrapper namespace can't drift. Generated content is spliced into two
pre-marked autogen sections in checked-in source files:
* ``src/ballistica/base/base.h`` — between the
``// __AUTOGENERATED_BUILTIN_ASSET_IDS_BEGIN__`` and ``…_END__``
markers: the four ``BuiltinTextureID`` / ``BuiltinCubeMapTextureID``
/ ``BuiltinSoundID`` / ``BuiltinMeshID`` enums + the
``kBuiltinAssetsApverid`` string constant.
* ``src/ballistica/base/assets/assets.cc`` — between the
``// __AUTOGENERATED_BUILTIN_ASSET_LOAD_BEGIN__`` and ``…_END__``
markers inside ``Assets::StartLoading()``: one
``LoadBuiltinTexture(BuiltinTextureID::kFooBar, "<apverid>:foo/bar")``
call per entry.
This is invoked by ``assetpins`` when the pin changes (it talks to the
cloud for the listing, so it lives there, not in ``make update`` --
which only *checks* the splice matches the pin and stays offline; see
``check_builtin_asset_ids``). The autogen sections live in checked-in
files. Idempotent — only writes a target file if the spliced result
differs from what's on disk, so a no-op regen leaves both files (and
their mtimes) untouched.
"""
from __future__ import annotations # Docs-generation hack.
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import TYPE_CHECKING
from efro.error import CleanError
if TYPE_CHECKING:
from pathlib import Path
[docs]
class AssetKind(Enum):
"""Which of the four C++ enums an entry belongs to."""
TEXTURE = 'texture'
CUBE_MAP_TEXTURE = 'cube_map_texture'
SOUND = 'sound'
MESH = 'mesh'
@property
def cpp_enum_name(self) -> str:
"""C++ enum class name for this kind."""
return {
AssetKind.TEXTURE: 'BuiltinTextureID',
AssetKind.CUBE_MAP_TEXTURE: 'BuiltinCubeMapTextureID',
AssetKind.SOUND: 'BuiltinSoundID',
AssetKind.MESH: 'BuiltinMeshID',
}[self]
@property
def cpp_loader_name(self) -> str:
"""C++ ``LoadBuiltin*`` function name for this kind."""
return {
AssetKind.TEXTURE: 'LoadBuiltinTexture',
AssetKind.CUBE_MAP_TEXTURE: 'LoadBuiltinCubeMapTexture',
AssetKind.SOUND: 'LoadBuiltinSound',
AssetKind.MESH: 'LoadBuiltinMesh',
}[self]
[docs]
@dataclass
class AssetEntry:
"""One asset, post-grouping & validation."""
kind: AssetKind
# Logical asset path within the package (workspace-relative, no
# extension). E.g. ``mydir/helloworld``.
logical_name: str
@property
def cpp_enum_entry(self) -> str:
"""``kMydirHelloworld`` form."""
return 'k' + ''.join(
_pascal_case(seg) for seg in self.logical_name.split('/')
)
[docs]
@dataclass
class BuildResult:
"""Output collected before writing to disk."""
apverid: str
entries: list[AssetEntry] = field(default_factory=list)
[docs]
def entries_for(self, kind: AssetKind) -> list[AssetEntry]:
"""Entries of a given asset kind, sorted by enum-entry name."""
return sorted(
(e for e in self.entries if e.kind == kind),
key=lambda e: e.cpp_enum_entry,
)
def _pascal_case(segment: str) -> str:
"""``some_thing`` → ``SomeThing``; ``foo-bar`` → invalid."""
if not re.fullmatch(r'[a-z0-9_]+', segment):
raise CleanError(
f'Asset-path segment {segment!r} is not lowercase '
'ascii letters/digits/underscores; rename in the workspace.'
)
return ''.join(part.capitalize() for part in segment.split('_'))
def _fetch_listing(projroot: Path, apverid: str) -> list[dict[str, str]]:
"""Fetch the assembly-free asset listing for ``apverid`` via bacloud.
Returns the ``assets`` list of ``{'path', 'kind'}`` dicts (see the
bamaster ``assetpackage _listing`` command). This is the one
cloud-talking step in builtin-id generation; it is invoked only from
``assetpins`` / ``gen_builtin_asset_ids`` (never ``update_project``,
which stays offline).
"""
import os
import json
import tempfile
import subprocess
tmpdir = projroot / 'build/tmp'
tmpdir.mkdir(parents=True, exist_ok=True)
fd, tmppath = tempfile.mkstemp(suffix='.json', dir=tmpdir)
os.close(fd)
try:
# Capture output so bacloud's own "Wrote asset listing..."
# end-message stays out of the (deliberately terse) assetpins
# output; surface it only if the call fails.
result = subprocess.run(
[
str(projroot / 'tools/bacloud'),
'assetpackage',
'_listing',
apverid,
tmppath,
],
check=True,
capture_output=True,
text=True,
)
del result # Output intentionally discarded on success.
with open(tmppath, encoding='utf-8') as infile:
data = json.load(infile)
assets = data['assets']
assert isinstance(assets, list)
return assets
except Exception as exc:
detail = ''
if isinstance(exc, subprocess.CalledProcessError):
msg = (exc.stderr or exc.stdout or '').strip()
if msg:
detail = f'\n{msg}'
raise CleanError(
f'Failed to fetch asset listing for {apverid!r}. If the'
f' apverid no longer exists on master (dev snapshots get'
f' pruned quickly), run `make assetpins-latest`.{detail}'
) from exc
finally:
if os.path.exists(tmppath):
os.unlink(tmppath)
[docs]
def collect(projroot: Path, apverid: str) -> BuildResult:
"""Fetch the asset listing for ``apverid`` and build a validated result.
The apverid is passed explicitly (not read from projectconfig) so a
caller staging a not-yet-written pin update can generate against the
*target* version. Fetches that version's asset listing via
``bacloud`` (assembly-free; see :func:`!_fetch_listing`). ``collect``
therefore talks to the cloud and must only run from ``assetpins`` /
``gen_builtin_asset_ids`` -- never ``update_project`` (which stays
offline and merely *checks* the spliced apverid string).
"""
assets = _fetch_listing(projroot, apverid)
result = BuildResult(apverid=apverid)
errors: list[str] = []
for asset in assets:
logical_name = asset['path']
# Kinds with no C++ id enum (collision_mesh today, plus any
# future kind the listing gains) have no AssetKind member;
# skip them. Keeps the generator forward-compatible without a
# listing-format gate.
try:
kind = AssetKind(asset['kind'])
except ValueError:
continue
segments = logical_name.split('/')
if len(segments) < 2:
errors.append(
f'Asset {logical_name!r} is at workspace root; '
'move into a category subdir (e.g. ui/, test/).'
)
continue
try:
for seg in segments:
_pascal_case(seg) # validation only
except CleanError as exc:
errors.append(str(exc))
continue
result.entries.append(AssetEntry(kind=kind, logical_name=logical_name))
# Cross-kind collision check: same logical_name appearing under two
# AssetKinds is ambiguous since the wrapper namespace is flat.
by_name: dict[str, list[AssetEntry]] = {}
for entry in result.entries:
by_name.setdefault(entry.logical_name, []).append(entry)
for name, entries in by_name.items():
kinds = {e.kind for e in entries}
if len(kinds) > 1:
errors.append(
f'Logical name {name!r} appears across multiple asset '
f'types ({sorted(k.value for k in kinds)}); rename to '
'disambiguate.'
)
if errors:
raise CleanError(
'Asset-package validation failed:\n - ' + '\n - '.join(errors)
)
return result
[docs]
def render_enum_block(result: BuildResult) -> str:
"""Build the autogen-section content for ``base.h``.
Returns the lines that go between
``// __AUTOGENERATED_BUILTIN_ASSET_IDS_BEGIN__`` and
``// __AUTOGENERATED_BUILTIN_ASSET_IDS_END__`` in base.h
(the markers themselves are NOT included).
"""
lines: list[str] = [
'//',
'// Generated by ``tools/pcommand gen_builtin_asset_ids`` (run by',
'// ``assetpins`` when the construct asset-package pin in',
'// ``pconfig/projectconfig.json`` changes) from that pin. Do not edit',
'// by hand; rerun ``make assetpins-latest`` to regenerate. New',
'// per-asset entries land here as the workspace gains them; old',
'// hand-coded ``Builtin*OldID`` entries above retire one at a time as',
'// their callsites migrate.',
'',
(
'inline constexpr const char* kBuiltinAssetsApverid = '
f'"{result.apverid}";'
),
'',
]
for kind in AssetKind:
entries = result.entries_for(kind)
if not entries:
lines.append(f'enum class {kind.cpp_enum_name} : uint16_t {{}};')
else:
lines.append(f'enum class {kind.cpp_enum_name} : uint16_t {{')
# Comment with the logical name — it matches
# wrapper/compat-table keys.
# clang-format aligns each run of trailing comments to
# its longest member, so place the comment on its own
# line above the entry whenever the aligned trailing
# form could exceed cpplint's 80-col limit (long mesh
# names hit this).
entry_parts = [f' {e.cpp_enum_entry},' for e in entries]
align_col = max(len(p) for p in entry_parts) + 2
for entry, part in zip(entries, entry_parts):
comment = f'// {entry.logical_name}'
if align_col + len(comment) > 80:
lines.append(f' {comment}')
lines.append(part)
else:
lines.append(f'{part} {comment}')
lines.append('};')
lines.append('')
# Drop the trailing blank line so the closing marker sits flush.
if lines and lines[-1] == '':
lines.pop()
return '\n'.join(lines)
[docs]
def render_load_block(result: BuildResult) -> str:
"""Build the autogen-section content for ``assets.cc``.
Returns the lines that go between
``// __AUTOGENERATED_BUILTIN_ASSET_LOAD_BEGIN__`` and
``// __AUTOGENERATED_BUILTIN_ASSET_LOAD_END__`` inside
``Assets::StartLoading()`` (the markers themselves are NOT
included). Lines that would exceed the 80-char cpplint limit
wrap after the comma.
"""
lines: list[str] = []
for kind in AssetKind:
entries = result.entries_for(kind)
if not entries:
continue
lines.append(f' // {kind.value}s')
for entry in entries:
full = f'{result.apverid}:{entry.logical_name}'
single = (
f' {kind.cpp_loader_name}('
f'{kind.cpp_enum_name}::{entry.cpp_enum_entry}, '
f'"{full}");'
)
if len(single) <= 80:
lines.append(single)
else:
# Wrap at the comma.
indent = ' ' * (len(kind.cpp_loader_name) + 3)
lines.append(
f' {kind.cpp_loader_name}('
f'{kind.cpp_enum_name}::{entry.cpp_enum_entry},'
)
lines.append(f'{indent}"{full}");')
return '\n'.join(lines)
_BEGIN_MARKER_PREFIX = '// __AUTOGENERATED_'
_END_MARKER_PREFIX = '// __AUTOGENERATED_'
def _splice_autogen(
existing: str, begin_marker: str, end_marker: str, new_content: str
) -> str:
"""Replace content between begin/end markers in ``existing``.
Marker lines (and any indentation in front of them) are
preserved; only the content strictly between them is replaced.
Raises ``CleanError`` if either marker isn't found or they're
in the wrong order.
"""
lines = existing.split('\n')
begin_idx: int | None = None
end_idx: int | None = None
for i, line in enumerate(lines):
stripped = line.lstrip()
if stripped == begin_marker:
if begin_idx is not None:
raise CleanError(f'Duplicate begin marker {begin_marker!r}.')
begin_idx = i
elif stripped == end_marker:
if end_idx is not None:
raise CleanError(f'Duplicate end marker {end_marker!r}.')
end_idx = i
if begin_idx is None:
raise CleanError(f'Begin marker {begin_marker!r} not found.')
if end_idx is None:
raise CleanError(f'End marker {end_marker!r} not found.')
if end_idx <= begin_idx:
raise CleanError(
f'End marker {end_marker!r} must come after begin '
f'marker {begin_marker!r}.'
)
new_lines = (
lines[: begin_idx + 1]
+ ([new_content] if new_content else [])
+ lines[end_idx:]
)
return '\n'.join(new_lines)
# Project-relative paths of the files that hold autogen sections.
TARGET_BASE_H = 'src/ballistica/base/base.h'
TARGET_ASSETS_CC = 'src/ballistica/base/assets/assets.cc'
_MARKERS_BASE_H = (
'// __AUTOGENERATED_BUILTIN_ASSET_IDS_BEGIN__',
'// __AUTOGENERATED_BUILTIN_ASSET_IDS_END__',
)
_MARKERS_ASSETS_CC = (
'// __AUTOGENERATED_BUILTIN_ASSET_LOAD_BEGIN__',
'// __AUTOGENERATED_BUILTIN_ASSET_LOAD_END__',
)
[docs]
def compute_splices(
projroot: Path,
apverid: str,
base_h_existing: str | None = None,
assets_cc_existing: str | None = None,
) -> dict[str, str]:
"""Compute spliced contents for both target files.
Returns a dict keyed by project-relative path with the full new
file content (existing content + new autogen section). Caller
decides whether to write — typical use is "write only if
contents differ from on-disk".
``base_h_existing`` / ``assets_cc_existing`` let the caller pass
in already-read content (e.g. when integrated into a project
updater that's already loaded the file); pass ``None`` to read
from disk here.
"""
# Run the spliced result through clang-format so our output matches
# what ``make format`` produces. Without this the generator emits
# raw (e.g. single-line) content that the formatter then rewrites
# (e.g. wrapping a long apverid assignment), leaving the two in a
# tug-of-war and making any "is the splice up to date?" check
# unreliable. The non-autogen parts of these files are already
# clang-formatted, so this only normalizes the spliced region.
from efrotools.code import format_cpp_str
result = collect(projroot, apverid)
if base_h_existing is None:
base_h_existing = (projroot / TARGET_BASE_H).read_text()
if assets_cc_existing is None:
assets_cc_existing = (projroot / TARGET_ASSETS_CC).read_text()
return {
TARGET_BASE_H: format_cpp_str(
projroot,
_splice_autogen(
base_h_existing,
_MARKERS_BASE_H[0],
_MARKERS_BASE_H[1],
render_enum_block(result),
),
filename='base.h',
),
TARGET_ASSETS_CC: format_cpp_str(
projroot,
_splice_autogen(
assets_cc_existing,
_MARKERS_ASSETS_CC[0],
_MARKERS_ASSETS_CC[1],
render_load_block(result),
),
filename='assets.cc',
),
}
[docs]
def generate(projroot: Path, apverid: str, check: bool = False) -> bool:
"""Splice generated content for ``apverid`` into base.h / assets.cc.
Reads each target file, replaces the content between its
``// __AUTOGENERATED_*__`` marker pair, and writes the file
only if the resulting content differs from what's on disk.
Idempotent: a run with no changes leaves both files (and their
mtimes) untouched.
Returns True if anything was (or would be) changed. (``assetpins``
stages writes itself via :func:`compute_splices`; this convenience
wrapper is used by the standalone ``gen_builtin_asset_ids``
pcommand.)
"""
spliced = compute_splices(projroot, apverid)
changed = False
for rel_path, new_text in spliced.items():
path = projroot / rel_path
existing = path.read_text()
if new_text == existing:
continue
changed = True
if check:
continue
path.write_text(new_text)
return changed
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum