OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-tps / scripts / gate_undersea_bridge.py

190 lines · python ref:
#!/usr/bin/env python3
"""GATE — the undersea graph in a REAL blender-bridge session (demo sprint §4).

The scene blender/undersea_geiss.tscn is the source of truth, edited in Godot or in
Blender through the bridge. This drives the session "Open in Blender" starts:
    Godot   scenes/test/undersea_bridge_driver.gd — the scene loaded, a bridge on a
            private socket, expect_godot_led(scene)
    Blender the addon's own _godot_launch_now(new_file=True) — the code the editor's
            launch runs — into a scratch .blend
and asserts:
    B1  the bridge pushed the scene on the bind: the Undersea object arrives as a
        DENSITY PREVIEW (blender-bridge wraps a density graph) with its 22 knobs
    B2  with seabed bounds the preview meshes, in the seabed's altitude band
    B3  a knob tuned ON ITS BLENDER NODE reaches Godot's live graph, the rest unchanged

Needs Blender 5.2 + the blender-bridge addon; SKIPs without them.
Run: blender --background --factory-startup --python scripts/gate_undersea_bridge.py
"""
import json
import os
import subprocess
import sys
import tempfile
import time

import bpy

HERE = os.path.dirname(os.path.abspath(__file__))
PROJECT = os.path.dirname(HERE)
GODOT = os.environ.get(
    "HYPER_TPS_GODOT",
    "/storage/projects/code/godot/nvidia-rtx-godot-v3/bin/godot.linuxbsd.editor.x86_64")
SCENE = "res://blender/undersea_geiss.tscn"
ADDON = "bl_ext.user_default.blender_godot_bridge"
KNOB = "floor_strength"
TUNED = 0.55

passed = 0
failed = 0


def ck(name, ok, detail=""):
    global passed, failed
    print("  %-62s %s  %s" % (name, "PASS" if ok else "FAIL", detail))
    if ok:
        passed += 1
    else:
        failed += 1


def main():
    if not os.path.exists(GODOT):
        print("SKIP: no Godot at %s" % GODOT)
        return 0
    try:
        bpy.ops.preferences.addon_enable(module=ADDON)
        bridge = sys.modules[ADDON]
    except Exception as e:
        print("SKIP: blender-bridge addon not available (%s)" % e)
        return 0
    gs = bridge.geometry_serializer
    scratch = tempfile.mkdtemp(prefix="undersea_bridge_")
    sock = os.path.join(scratch, "b.sock")
    shm = "/undersea_" + os.path.basename(scratch)
    env = dict(os.environ, CI="true", HYPERTPS_DAEMON_TCP="127.0.0.1:1")
    log = open(os.path.join(scratch, "godot.log"), "w")
    proc = subprocess.Popen(
        [GODOT, "--headless", "--path", PROJECT, "--script",
         "res://scenes/test/undersea_bridge_driver.gd", "--", scratch, sock, shm],
        env=env, stdout=log, stderr=subprocess.STDOUT)
    gen = [0]

    def pump(n=5, delay=0.03):
        for _ in range(n):
            bridge._poll_timer_impl(gen[0])
            time.sleep(delay)

    def report(tag, timeout=30.0):
        open(os.path.join(scratch, "cmd_report"), "w").write(tag)
        deadline = time.time() + timeout
        while time.time() < deadline:
            if os.path.exists(os.path.join(scratch, "ack_report_" + tag)):
                with open(os.path.join(scratch, "undersea_state.json")) as f:
                    return json.load(f)
            pump(2)
        return None

    try:
        deadline = time.time() + 120
        while time.time() < deadline and not os.path.exists(os.path.join(scratch, "godot_ready")):
            time.sleep(0.1)
        ck("the Godot driver came up", os.path.exists(os.path.join(scratch, "godot_ready")))
        seed = report("seed")
        ck("Godot holds the scene's graph (%s knobs)" % (len(seed["knobs"]) if seed else "?"),
           seed is not None and len(seed["knobs"]) == 22)
        if seed is None:
            return 1

        res = bridge._godot_launch_now(sock, shm, SCENE, os.path.join(scratch, "undersea.blend"), True)
        ck("the addon connected (Godot-led launch)", "FINISHED" in res, repr(res))
        gen[0] = bridge._timer_generation
        obj = None
        deadline = time.time() + 60
        while time.time() < deadline:
            pump(5)
            obj = bpy.data.objects.get("Undersea")
            mod = next((m for m in obj.modifiers if m.type == "NODES"), None) if obj else None
            if mod is not None and mod.node_group is not None:
                break
        mod = next((m for m in obj.modifiers if m.type == "NODES"), None) if obj else None
        tree, _ = gs.graph_of(mod) if mod else (None, None)
        ck("B1 the Undersea object arrived as a density preview",
           mod is not None and bool(mod.node_group.get(gs.PREVIEW_TAG)) and tree is not None,
           sorted(o.name for o in bpy.data.objects))
        if tree is None:
            return 1
        knobs = {n.label[5:]: n for n in tree.nodes if n.label.startswith("knob:")}
        ck("B1 its 22 knobs are labelled Blender nodes", len(knobs) == 22, len(knobs))

        # Godot (x, y, z) shows as (-x, z, y): the seabed's altitude is Blender Z.
        ids ={i.name: i.identifier for i in mod.node_group.interface.items_tree
               if getattr(i, "in_out", "") == "INPUT"}
        # Blender 5.2 keeps modifier inputs as RNA (geometry_serializer.modifier_input
        # reads the same place); mod[identifier] is refused.
        for name, value in (("Min", (-150.0, -150.0, -260.0)), ("Max", (150.0, 150.0, 0.0)),
                            ("Resolution", 96)):
            getattr(mod.properties.inputs, ids[name]).value = value
        obj.update_tag()
        bpy.context.view_layer.update()
        ev = obj.evaluated_get(bpy.context.evaluated_depsgraph_get())
        me = ev.to_mesh()
        zs = [v.co.z for v in me.vertices]
        ev.to_mesh_clear()
        # Volume to Mesh closes the surface up to ONE VOXEL outside the box (260 m / 96
        # cells = 2.7 m in z); measured 51734 verts, z -262.7 .. -113.9 — the same as the
        # P4 harness measured before the preview moved into the bridge.
        voxel = 260.0 / 96
        ck("B2 the preview meshes the seabed (z in -260..0, ± one voxel)",
           len(zs) > 1000 and min(zs) >= -260.0 - voxel - 0.1 and max(zs) <= voxel + 0.1,
           "%d verts, z %.1f .. %.1f" % (len(zs), min(zs) if zs else 0, max(zs) if zs else 0))

        knobs[KNOB].outputs[0].default_value = TUNED
        bpy.context.view_layer.update()
        got = None
        deadline = time.time() + 30
        while time.time() < deadline:
            pump(5)
            st = report("t%d" % int(time.time() * 1000))
            if st and abs(float(st["knobs"][KNOB]) - TUNED) < 1e-4:
                got = st
                break
        ck("B3 the knob tuned in Blender reached Godot's live graph", got is not None,
           got["knobs"][KNOB] if got else (st["knobs"][KNOB] if st else "no report"))
        if got:
            before = {k: v for k, v in seed["knobs"].items() if k != KNOB}
            diff = _knob_diff(before, got["knobs"])
            ck("B3 every other knob is unchanged", not diff, diff)
    finally:
        open(os.path.join(scratch, "cmd_quit"), "w").write("1")
        try:
            proc.wait(timeout=30)
        except subprocess.TimeoutExpired:
            proc.kill()
        log.close()
        with open(os.path.join(scratch, "godot.log")) as f:
            tail = [l for l in f.read().splitlines() if "[BB]" in l or "SCRIPT ERROR" in l][-8:]
        print("  godot:\n    " + "\n    ".join(tail))

    print("%d/%d passed" % (passed, passed + failed))
    return 0 if failed == 0 else 1


def _knob_diff(a, b):
    """Keys of `a` whose knob differs in `b`. Knobs arrive as var_to_str text: a float
    ("0.9") or a Vector3 ("Vector3(0.61, 1.13, 0.37)"), compared component-wise."""
    out = []
    for k, v in a.items():
        w = b.get(k)
        if w is None:
            out.append(k)
            continue
        pa = [float(x) for x in str(v).replace("Vector3(", "").rstrip(")").split(",")]
        pb = [float(x) for x in str(w).replace("Vector3(", "").rstrip(")").split(",")]
        if len(pa) != len(pb) or any(abs(x - y) > 1e-4 for x, y in zip(pa, pb)):
            out.append(k)
    return sorted(out)


sys.exit(main())