vkgs-sim / scripts /shader_graph.py
kimjunhan's picture
Add files using upload-large-folder tool
4c75661 verified
Raw History Blame Contribute Delete
10.1 kB
#!/usr/bin/env python3
"""셰이더 κ·Έλž˜ν”„ β†’ GLSL μ»΄ν“¨νŠΈ 셰이더 μ½”λ“œ 생성.
κ·Έλž˜ν”„(configs/shader_graphs/<이름>.json)λŠ” λ…Έλ“œμ™€ 간선이닀. λ…Έλ“œμ˜ μ—°μ‚°(op)은 λ…Έλ“œ 라이브러리
(src/vkgs/shaders/nodes/*.glsl)의 ν•¨μˆ˜ n_<op> 이고, κ·Έ **μ‹œκ·Έλ‹ˆμ²˜λ₯Ό 읽어** μž…λ ₯ ν¬νŠΈΒ·λ§€κ°œλ³€μˆ˜Β·νƒ€μž…μ„ μ–»λŠ”λ‹€
(라이브러리 μͺ½ 'λ¦¬ν”Œλ ‰μ…˜' β€” λ…Έλ“œλ₯Ό 더할 λ•Œ 이 μŠ€ν¬λ¦½νŠΈλŠ” κ³ μΉ˜μ§€ μ•ŠλŠ”λ‹€). 특수 λ…Έλ“œ 두 개:
input β€” 이미지 픽셀을 μ½λŠ”λ‹€ (vec3)
output β€” μž…λ ₯ ν•˜λ‚˜λ₯Ό 이미지에 μ“΄λ‹€
λ…Έλ“œλ₯Ό μœ„μƒ μ •λ ¬ν•΄ ν•œ 쀄씩 ν•¨μˆ˜ 호좜둜 펼치고, κ°„μ„  νƒ€μž…μ΄ λ§žλŠ”μ§€Β·μˆœν™˜Β·λΉ μ§„ μž…λ ₯을 κ²€μ‚¬ν•œλ‹€.
κ²°κ³Ό (OUT 디렉터리):
<이름>.comp λ Œλ” κ·Έλž˜ν”„μ˜ compute νŒ¨μŠ€κ°€ μ“°λŠ” 셰이더 (CMake κ°€ SPIR-V 둜 κ΅½λŠ”λ‹€)
<이름>.params.json λ§€κ°œλ³€μˆ˜ ν‘œ (이름·버퍼 μœ„μΉ˜Β·κΈ°λ³Έκ°’Β·λ²”μœ„) β€” λ·°μ–΄κ°€ μŠ¬λΌμ΄λ”μ™€ μ΄ˆκΉƒκ°’μ— μ“΄λ‹€
<이름>.dot κ·Έλ¦Ό (dot -Tsvg)
μ‚¬μš©: shader_graph.py GRAPH.json --shaders src/vkgs/shaders --out build/shaders
μ‹œμŠ€ν…œ 파이썬(3.8)μ—μ„œλ„ λˆλ‹€ β€” ν‘œμ€€ 라이브러리만 μ“΄λ‹€.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path
SPECIAL = ("input", "output")
_FUNC = re.compile(r"^\s*(\w+)\s+n_(\w+)\s*\(([^)]*)\)", re.M)
_PARAM_DOC = re.compile(r"^\s*//\s*@param\s+(\w+)\s+(\S+)\s+(\S+)\s+(\S+)")
class GraphError(Exception):
pass
def parse_library(text: str) -> dict:
"""λ…Έλ“œ 라이브러리 GLSL β†’ {op: {ret, inputs: [(이름, νƒ€μž…)], ctx, params: [(이름, κΈ°λ³Έ, μ΅œμ†Œ, μ΅œλŒ€)]}}."""
lines = text.splitlines()
ops = {}
for m in _FUNC.finditer(text):
ret, op, args = m.group(1), m.group(2), m.group(3)
if ret == "return": # λ³Έλ¬Έ μ•ˆμ˜ 호좜 (return n_x(...))
continue
line_no = text.count("\n", 0, m.start())
# λ°”λ‘œ μœ„ 주석 λΈ”λ‘μ˜ @param
docs = {}
k = line_no - 1
while k >= 0 and lines[k].strip().startswith("//"):
d = _PARAM_DOC.match(lines[k])
if d:
docs[d.group(1)] = tuple(float(d.group(i)) for i in (2, 3, 4))
k -= 1
inputs, params, ctx = [], [], False
for a in [s.strip() for s in args.split(",") if s.strip()]:
parts = a.split()
if len(parts) != 2:
raise GraphError(f"λ…Έλ“œ n_{op}: 인자λ₯Ό 읽을 수 μ—†μŒ '{a}'")
typ, name = parts
if typ == "Ctx":
ctx = True
elif ctx:
if typ != "float":
raise GraphError(f"λ…Έλ“œ n_{op}: λ§€κ°œλ³€μˆ˜ {name} λŠ” float μ—¬μ•Ό 함 ({typ})")
params.append((name,) + docs.get(name, (0.0, 0.0, 1.0)))
else:
inputs.append((name, typ))
if op in ops:
raise GraphError(f"λ…Έλ“œ n_{op} κ°€ 두 번 μ •μ˜λ¨")
ops[op] = {"ret": ret, "inputs": inputs, "ctx": ctx, "params": params}
return ops
def _topo(nodes: dict) -> list:
"""Kahn β€” 같은 λ‹¨κ³„λŠ” μ„ μ–Έ μˆœμ„œ. μˆœν™˜μ΄λ©΄ GraphError."""
ids = list(nodes)
indeg = {i: 0 for i in ids}
outs = {i: [] for i in ids}
for i, n in nodes.items():
for src in n.get("in", []):
if src not in nodes:
raise GraphError(f"λ…Έλ“œ {i}: μ—†λŠ” λ…Έλ“œ '{src}' λ₯Ό μž…λ ₯으둜 씀")
outs[src].append(i)
indeg[i] += 1
ready = [i for i in ids if indeg[i] == 0]
order = []
while ready:
ready.sort(key=ids.index)
i = ready.pop(0)
order.append(i)
for j in outs[i]:
indeg[j] -= 1
if indeg[j] == 0:
ready.append(j)
if len(order) != len(ids):
raise GraphError("셰이더 κ·Έλž˜ν”„μ— μˆœν™˜μ΄ 있음: " + ", ".join(i for i in ids if i not in order))
return order
def generate(graph: dict, lib_ops: dict, name: str, includes: list) -> tuple:
"""κ·Έλž˜ν”„ β†’ (GLSL μ†ŒμŠ€, λ§€κ°œλ³€μˆ˜ ν‘œ, DOT)."""
nodes = {k: v for k, v in graph["nodes"].items() if not k.startswith("_")}
target = graph.get("target", {})
image = target.get("image", "kImgPost")
order = _topo(nodes)
types, params, body = {}, [], []
n_out = 0
for i in order:
n = nodes[i]
op = n.get("op")
ins = n.get("in", [])
if op == "input":
if ins:
raise GraphError(f"λ…Έλ“œ {i}: input 은 μž…λ ₯이 μ—†λ‹€")
img = n.get("image", image)
body.append(f" vec3 v_{i} = imageLoad(gImgF4[{img}], x.px).rgb;")
types[i] = "vec3"
continue
if op == "output":
if len(ins) != 1 or types[ins[0]] != "vec3":
raise GraphError(f"λ…Έλ“œ {i}: output 은 vec3 μž…λ ₯ ν•˜λ‚˜")
img = n.get("image", image)
body.append(f" imageStore(gImgF4[{img}], x.px, vec4(v_{ins[0]}, imageLoad(gImgF4[{img}], x.px).a));")
n_out += 1
continue
if op not in lib_ops:
raise GraphError(f"λ…Έλ“œ {i}: λΌμ΄λΈŒλŸ¬λ¦¬μ— μ—†λŠ” μ—°μ‚° '{op}' (μžˆλŠ” 것: {', '.join(sorted(lib_ops))})")
spec = lib_ops[op]
if len(ins) != len(spec["inputs"]):
raise GraphError(f"λ…Έλ“œ {i} ({op}): μž…λ ₯ {len(spec['inputs'])}κ°œκ°€ ν•„μš”ν•œλ° {len(ins)}개")
for src, (port, typ) in zip(ins, spec["inputs"]):
if types[src] != typ:
raise GraphError(f"λ…Έλ“œ {i} ({op}): μž…λ ₯ {port} λŠ” {typ} 인데 {src} λŠ” {types[src]}")
given = n.get("params", {})
unknown = set(given) - {p[0] for p in spec["params"]}
if unknown:
raise GraphError(f"λ…Έλ“œ {i} ({op}): λͺ¨λ₯΄λŠ” λ§€κ°œλ³€μˆ˜ {sorted(unknown)}")
args = [f"v_{s}" for s in ins]
if spec["ctx"]:
args.append("x")
for pname, default, lo, hi in spec["params"]:
idx = len(params)
params.append({"name": f"{i}.{pname}", "node": i, "op": op, "index": idx,
"default": float(given.get(pname, default)), "min": lo, "max": hi})
args.append(f"P({idx})")
body.append(f" {spec['ret']} v_{i} = n_{op}({', '.join(args)});")
types[i] = spec["ret"]
if n_out == 0:
raise GraphError("output λ…Έλ“œκ°€ μ—†μŒ")
inc = "\n".join(f'#include "{p}"' for p in includes)
src = f"""#version 450
#extension GL_GOOGLE_include_directive : require
#extension GL_EXT_shader_image_load_formatted : require
// μžλ™ 생성 β€” 셰이더 κ·Έλž˜ν”„ '{name}' (scripts/shader_graph.py). κ³ μΉ˜μ§€ 말고 κ·Έλž˜ν”„ JSONΒ·λ…Έλ“œ 라이브러리λ₯Ό κ³ μΉ  것.
#include "bindless.glsl"
layout(local_size_x = 8, local_size_y = 8) in;
layout(set = 0, binding = 4) uniform image2D gImgF4[]; // ν˜•μ‹ μ—†μŒ β€” λ Œλ” κ·Έλž˜ν”„ JSON 이 ν˜•μ‹μ„ μ •ν•œλ‹€
layout(std430, set = 0, binding = 0) readonly buffer GraphParams {{ float v[]; }} gGraphParams[];
// slots.x = λ§€κ°œλ³€μˆ˜ λ²„νΌμ˜ SSBO 슬둯, ctx = (μ‹œκ° s, ν”„λ ˆμž„ 번호 λΉ„νŠΈ)
layout(push_constant) uniform Push {{ uvec4 slots; vec4 ctx; }} pc;
#define P(i) gGraphParams[pc.slots.x].v[i]
{inc}
void main() {{
ivec2 size = imageSize(gImgF4[{image}]);
Ctx x;
x.px = ivec2(gl_GlobalInvocationID.xy);
if (x.px.x >= size.x || x.px.y >= size.y) return;
x.size = vec2(size);
x.uv = (vec2(x.px) + 0.5) / x.size;
x.time = pc.ctx.x;
x.frame = floatBitsToUint(pc.ctx.y);
{chr(10).join(body)}
}}
"""
table = {"graph": name, "resource": target.get("params_resource"), "count": len(params), "params": params}
return src, table, to_dot(name, nodes, order, lib_ops)
def to_dot(name: str, nodes: dict, order: list, lib_ops: dict) -> str:
out = [f'digraph "{name}" {{', " rankdir=LR;", ' node [fontname="NanumGothic", shape=record, style=filled];']
for i in order:
n = nodes[i]
op = n.get("op")
spec = lib_ops.get(op, {"params": []})
ps = "\\l".join(f"{p[0]} = {n.get('params', {}).get(p[0], p[1]):g}" for p in spec["params"])
color = "#d5f5e3" if op in SPECIAL else "#fde8c8"
label = f"{i}\\n[{op}]" + (f"|{ps}\\l" if ps else "")
out.append(f' "{i}" [label="{{{label}}}", fillcolor="{color}"];')
for s in n.get("in", []):
out.append(f' "{s}" -> "{i}";')
out.append("}")
return "\n".join(out) + "\n"
def build(graph_path: Path, shaders: Path, out: Path) -> Path:
graph = json.loads(graph_path.read_text())
name = graph.get("name", graph_path.stem)
includes = graph.get("library", ["nodes/isp.glsl"])
lib_ops = {}
for inc in includes:
for op, spec in parse_library((shaders / inc).read_text()).items():
if op in lib_ops:
raise GraphError(f"λ…Έλ“œ n_{op} κ°€ 라이브러리 μ—¬λŸ¬ 곳에 있음")
lib_ops[op] = spec
src, table, dot = generate(graph, lib_ops, name, includes)
out.mkdir(parents=True, exist_ok=True)
comp = out / f"{name}.comp"
# λ‚΄μš©μ΄ κ°™μœΌλ©΄ μ“°μ§€ μ•ŠλŠ”λ‹€ (λΆˆν•„μš”ν•œ 재컴파일 λ°©μ§€)
if not comp.exists() or comp.read_text() != src:
comp.write_text(src)
(out / f"{name}.params.json").write_text(json.dumps(table, indent=2, ensure_ascii=False) + "\n")
(out / f"{name}.dot").write_text(dot)
return comp
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
ap.add_argument("graph", type=Path)
ap.add_argument("--shaders", type=Path, required=True, help="λ…Έλ“œ 라이브러리 κΈ°μ€€ 디렉터리 (src/vkgs/shaders)")
ap.add_argument("--out", type=Path, required=True)
a = ap.parse_args(argv)
try:
comp = build(a.graph, a.shaders, a.out)
except GraphError as e:
print(f"셰이더 κ·Έλž˜ν”„ 였λ₯˜ ({a.graph}): {e}", file=sys.stderr)
return 1
print(f"셰이더 κ·Έλž˜ν”„ {a.graph.name} β†’ {comp}")
return 0
if __name__ == "__main__":
sys.exit(main())