158 lines
8.2 KiB
Python
158 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Compile Slang and atomically publish SPIR-V with Faset reflection v1.
|
|
|
|
The layout fingerprint contains the declared shader interface and actual SPIR-V
|
|
push-constant decorations, never compiler formatting or generated ID numbers.
|
|
A failed compiler invocation or reflection conversion leaves previous files intact.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
def digest(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def canonical(value: object) -> bytes:
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def value_type(value: dict) -> str:
|
|
kind = value["kind"]
|
|
if kind == "scalar":
|
|
return value["scalarType"]
|
|
if kind == "vector":
|
|
return f"{value_type(value['elementType'])}x{value['elementCount']}"
|
|
if kind == "matrix":
|
|
return f"{value_type(value['elementType'])}x{value['rowCount']}x{value['columnCount']}"
|
|
if kind == "array":
|
|
return f"{value_type(value['elementType'])}[{value['elementCount']}]"
|
|
raise ValueError(f"Unsupported reflected value type: {kind}")
|
|
|
|
|
|
def interface(fields: list[dict], category: str) -> tuple[list[dict], list[dict]]:
|
|
locations, builtins = [], []
|
|
def visit(field: dict, offset: int = 0) -> None:
|
|
ty, binding = field["type"], field.get("binding", {})
|
|
if ty["kind"] == "struct":
|
|
for child in ty["fields"]:
|
|
visit(child, offset + binding.get("index", 0))
|
|
elif binding.get("kind") == category:
|
|
locations.append({"location": offset + binding["index"], "type": value_type(ty)})
|
|
elif field.get("semanticName", "").startswith("SV_"):
|
|
builtins.append({"semantic": field["semanticName"], "type": value_type(ty)})
|
|
else:
|
|
raise ValueError("Shader interface field lacks a supported location or builtin")
|
|
for field in fields:
|
|
visit(field)
|
|
return sorted(locations, key=lambda item: item["location"]), sorted(builtins, key=lambda item: item["semantic"])
|
|
|
|
|
|
def spirv_push_layout(data: bytes) -> list[dict]:
|
|
if len(data) < 20 or len(data) % 4:
|
|
raise ValueError("Malformed SPIR-V byte length")
|
|
words = struct.unpack(f"<{len(data) // 4}I", data)
|
|
if words[0] != 0x07230203 or words[1] > 0x00010600 or not words[3] or words[4]:
|
|
raise ValueError("Unsupported SPIR-V header")
|
|
pointers, variables, decorations = {}, [], {}
|
|
position = 5
|
|
while position < len(words):
|
|
count, opcode = words[position] >> 16, words[position] & 0xFFFF
|
|
if not count or position + count > len(words):
|
|
raise ValueError("Malformed SPIR-V instruction")
|
|
operands = words[position + 1:position + count]
|
|
if opcode == 32 and len(operands) == 3: # OpTypePointer
|
|
pointers[operands[0]] = (operands[1], operands[2])
|
|
elif opcode == 59 and len(operands) >= 3 and operands[2] == 9: # PushConstant OpVariable
|
|
variables.append(operands[0])
|
|
elif opcode == 72 and len(operands) >= 3: # OpMemberDecorate
|
|
member = decorations.setdefault(operands[0], {}).setdefault(operands[1], {})
|
|
if operands[2] in (4, 5):
|
|
member["matrix_layout"] = "row-major" if operands[2] == 4 else "column-major"
|
|
elif operands[2] in (7, 35):
|
|
if len(operands) != 4:
|
|
raise ValueError("Malformed SPIR-V member decoration")
|
|
member["matrix_stride" if operands[2] == 7 else "offset"] = operands[3]
|
|
position += count
|
|
result = []
|
|
for pointer in variables:
|
|
storage, structure = pointers[pointer]
|
|
if storage != 9:
|
|
raise ValueError("Invalid SPIR-V push-constant pointer")
|
|
members = [{"member": index, **layout} for index, layout in sorted(decorations.get(structure, {}).items())]
|
|
result.append({"members": members})
|
|
return result
|
|
|
|
|
|
def normalize(raw: dict, bytecode: bytes, entry_name: str) -> dict:
|
|
entry = next(item for item in raw["entryPoints"] if item["name"] == entry_name)
|
|
used = {item["name"]: bool(item["binding"].get("used", True)) for item in entry.get("bindings", [])}
|
|
descriptors, constants = [], []
|
|
for parameter in raw.get("parameters", []):
|
|
binding, ty = parameter["binding"], parameter["type"]
|
|
if binding["kind"] == "pushConstantBuffer":
|
|
block = ty["elementType"]
|
|
size = next(item["value"] for item in block["sizes"] if item["kind"] == "uniform")
|
|
members = [{"name": member["name"], "offset": member["binding"]["offset"], "size": member["binding"]["size"], "type": value_type(member["type"])} for member in block["fields"]]
|
|
constants.append({"name": parameter["name"], "offset": 0, "size": size, "members": sorted(members, key=lambda item: item["offset"])})
|
|
elif binding["kind"] == "descriptorTableSlot":
|
|
count = binding.get("count", 1)
|
|
if ty["kind"] == "array":
|
|
count = ty["elementCount"]
|
|
ty = ty["elementType"]
|
|
if ty["kind"] == "samplerState":
|
|
descriptor_type = "sampler"
|
|
elif ty["kind"] == "resource" and ty.get("baseShape") == "texture2D":
|
|
descriptor_type = "sampled_image_2d"
|
|
else:
|
|
raise ValueError(f"Unsupported descriptor kind: {ty}")
|
|
descriptors.append({"name": parameter["name"], "set": binding.get("space", 0), "binding": binding["index"], "type": descriptor_type, "count": count, "used": used.get(parameter["name"], True)})
|
|
else:
|
|
raise ValueError(f"Unsupported global shader binding: {binding['kind']}")
|
|
inputs, input_builtins = interface(entry.get("parameters", []), "varyingInput")
|
|
outputs, output_builtins = interface([entry["result"]] if "result" in entry else [], "varyingOutput")
|
|
layout = {"stage": entry["stage"], "descriptors": sorted(descriptors, key=lambda item: (item["set"], item["binding"])), "push_constants": constants, "inputs": inputs, "outputs": outputs, "input_builtins": input_builtins, "output_builtins": output_builtins, "spirv_push_constants": spirv_push_layout(bytecode)}
|
|
return {"format": "faset.shader-reflection", "version": 1, "source_entry": entry_name, "entry_point": "main", "matrix_convention": "column-major host matrices; Slang SPIR-V decorations recorded explicitly", "spirv_sha256": digest(bytecode), "layout_fingerprint": digest(canonical(layout)), "layout": layout}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--compiler", required=True)
|
|
parser.add_argument("--source", required=True, type=Path)
|
|
parser.add_argument("--entry", required=True)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
args.output.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix=".shader-", dir=args.output) as temporary:
|
|
directory = Path(temporary)
|
|
spirv = directory / f"{args.entry}.spv"
|
|
raw = directory / f"{args.entry}.slang-reflection.json"
|
|
process = subprocess.run([args.compiler, str(args.source), "-entry", args.entry, "-target", "spirv", "-profile", "spirv_1_6", "-matrix-layout-column-major", "-o", str(spirv), "-reflection-json", str(raw)])
|
|
if process.returncode:
|
|
return process.returncode
|
|
normalized = normalize(json.loads(raw.read_text(encoding="utf-8")), spirv.read_bytes(), args.entry)
|
|
manifest = directory / f"{args.entry}.reflection.json"
|
|
manifest.write_text(json.dumps(normalized, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
# Reflection is the commit record. A reader racing these replacements rejects
|
|
# a hash mismatch and keeps its existing pipelines until the complete pair arrives.
|
|
for artifact in (raw, spirv, manifest):
|
|
os.replace(artifact, args.output / artifact.name)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (OSError, ValueError, KeyError, StopIteration) as error:
|
|
print(f"Shader reflection error: {error}", file=sys.stderr)
|
|
raise SystemExit(1)
|