Download scripts/shader_graph.py from kimjunhan/vkgs-sim: direct link, hf CLI and curl.
- Browser
- Download file 10.1 kB
-
https://huggingface.co/kimjunhan/vkgs-sim/resolve/main/scripts/shader_graph.py
- Command line
-
hf download hf://kimjunhan/vkgs-sim/scripts/shader_graph.py
-
curl -L -o shader_graph.py https://huggingface.co/kimjunhan/vkgs-sim/resolve/main/scripts/shader_graph.py
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()) | |