animation/tools/blender_bridge/tinqs_blender_bridge.py

"""TinqsBlenderBridge -- live-control socket server for a running Blender.

Runs a localhost-only TCP server inside Blender and executes Python source
sent to it, giving external tooling (Claude Code) live control of the open
Blender session via bpy. Counterpart to the iClone TinqsBridge (port 18800)
and TinqsMDBridge (port 18900); this one serves on port 19000.

Unlike MD, Blender has bpy.app.timers: the server polls a NON-BLOCKING
socket from a main-thread timer, so the UI stays fully interactive while
the bridge runs. Each request's exec() happens on the main thread (the only
safe place for bpy calls); the UI stalls only for the duration of that one
exec.

Install / start:
  - One-off, in the open window: Scripting workspace > open this file >
    Run Script. The bridge starts immediately.
  - Auto-start on every launch: run tools/install_blender_bridge.ps1, which
    copies this file into Blender's scripts/startup/ folder (startup modules
    are imported at launch and their register() is called).
  - Verify from a terminal:  python tools/blender_bridge.py --ping
  - Stop:                    python tools/blender_bridge.py --stop

Wire protocol (identical to the iClone/MD bridges): newline-delimited JSON
over TCP, one request/response pair per connection.

    request:  {"id": <int>, "code": "<python source>"}
    success:  {"id": <int>, "ok": true,  "result": <json|null>, "stdout": "<str>"}
    failure:  {"id": <int>, "ok": false, "error": "<traceback str>", "stdout": "<str>"}

Execution semantics (same `result` convention as the other bridges):
  - exec(code, ns, ns) against ONE persistent namespace, pre-seeded with
    bpy, a BRIDGE info dict, and view3d_override() (see below). It survives
    across requests; re-running this file re-seeds it.
  - Setting `result` in submitted code makes it the response result
    (JSON-encoded, repr() fallback). Cleared before every exec.
  - stdout/stderr captured and returned; exceptions come back as ok:false
    with a traceback and the bridge keeps serving.

Context caveat: timer callbacks run with NO window/area in bpy.context, so
operators that need one (most bpy.ops.view3d.*, some object ops) fail with
"context is incorrect". Prefer the data API (bpy.data, obj.location, ...).
When an operator genuinely needs a 3D viewport, the seeded helper gives you
an override:

    with bpy.context.temp_override(**view3d_override()):
        bpy.ops.view3d.view_selected()

Log file: %TEMP%/tinqs_blender_bridge.log
"""

import builtins
import contextlib
import io
import json
import os
import socket
import traceback
from datetime import datetime

import bpy

VERSION = 1
DEFAULT_PORT = 19000
HOST = "127.0.0.1"
POLL_INTERVAL_S = 0.1
CONN_READ_TIMEOUT_S = 10.0
SINGLETON_ATTR = "_tinqs_blender_bridge"
STOP_SENTINEL = "__STOP__"

LOG_PATH = os.path.join(
    os.environ.get("TEMP", os.environ.get("TMP", ".")),
    "tinqs_blender_bridge.log",
)


def _log(msg):
    """Append a timestamped line to the log file. Must never raise."""
    try:
        line = "[{}] {}\n".format(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), msg)
        with open(LOG_PATH, "a", encoding="utf-8") as f:
            f.write(line)
    except Exception:
        pass


def _safe_blend_file():
    try:
        return bpy.data.filepath or "<unsaved>"
    except AttributeError:
        return "<unknown -- restricted context>"


def view3d_override():
    """Kwargs for bpy.context.temp_override() targeting the first 3D viewport."""
    wm = bpy.data.window_managers[0]
    for window in wm.windows:
        for area in window.screen.areas:
            if area.type == "VIEW_3D":
                region = next(r for r in area.regions if r.type == "WINDOW")
                return {"window": window, "area": area, "region": region}
    raise RuntimeError("no VIEW_3D area found in any open window")


class _Bridge:
    def __init__(self, port):
        self.port = port
        self.sock = None
        self.served = 0
        import sys
        self.ns = {
            "bpy": bpy,
            "view3d_override": view3d_override,
        }
        self.ns["BRIDGE"] = {
            "version": VERSION,
            "mode": "timer",
            "port": port,
            "source": os.path.abspath(__file__) if "__file__" in globals() else "<text editor>",
            "blender": bpy.app.version_string,
            "python": sys.version,
            # bpy.data is a restricted stub during startup-module registration
            # (AttributeError on .filepath); --ping refreshes this live anyway.
            "blend_file": _safe_blend_file(),
        }

    def start(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind((HOST, self.port))
        sock.listen(5)
        sock.setblocking(False)
        self.sock = sock
        bpy.app.timers.register(self._poll, persistent=True)
        _log("TinqsBlenderBridge v{} serving on {}:{} blender={}".format(
            VERSION, HOST, self.port, bpy.app.version_string))
        print("TinqsBlenderBridge v{}: serving on {}:{} (UI stays interactive)".format(
            VERSION, HOST, self.port))
        print("Verify:  python tools/blender_bridge.py --ping")

    def stop(self, in_timer=False):
        # Unregistering a timer from inside its own callback is an error in
        # Blender -- when stopping from _poll, closing the socket is enough;
        # the callback's `return None` retires the timer.
        if not in_timer and bpy.app.timers.is_registered(self._poll):
            bpy.app.timers.unregister(self._poll)
        if self.sock is not None:
            try:
                self.sock.close()
            except Exception:
                pass
            self.sock = None
        _log("bridge stopped, served {}".format(self.served))

    # -- timer callback: runs on the main thread every POLL_INTERVAL_S --
    def _poll(self):
        try:
            conn, _addr = self.sock.accept()
        except BlockingIOError:
            return POLL_INTERVAL_S
        except OSError:
            _log("server socket died:\n{}".format(traceback.format_exc()))
            self.stop(in_timer=True)
            return None
        try:
            self._handle(conn)
        except Exception:
            _log("connection error:\n{}".format(traceback.format_exc()))
        finally:
            try:
                conn.close()
            except Exception:
                pass
        if self.sock is None:  # _handle saw __STOP__
            return None
        return POLL_INTERVAL_S

    def _handle(self, conn):
        try:
            req = self._read_request(conn)
        except Exception:
            self._respond(conn, {"id": None, "ok": False,
                                 "error": "invalid JSON request", "stdout": ""})
            return
        if req is None:
            return
        code = req.get("code", "")
        if code.strip() == STOP_SENTINEL:
            self._respond(conn, {"id": req.get("id"), "ok": True,
                                 "result": "stopped", "stdout": ""})
            print("TinqsBlenderBridge: stopped by client (served {})".format(self.served))
            self.stop(in_timer=True)
            return
        resp = self._execute(code)
        resp["id"] = req.get("id")
        self.served += 1
        if not resp["ok"]:
            last = resp["error"].strip().splitlines()[-1] if resp["error"] else "?"
            _log("request id={} ok=false: {}".format(req.get("id"), last))
        self._respond(conn, resp)

    def _execute(self, code):
        self.ns["result"] = None
        out = io.StringIO()
        try:
            with contextlib.redirect_stdout(out), contextlib.redirect_stderr(out):
                exec(code, self.ns, self.ns)
            result = self.ns.get("result")
            try:
                json.dumps(result)
            except TypeError:
                result = repr(result)
            return {"ok": True, "result": result, "stdout": out.getvalue()}
        except Exception:
            return {"ok": False, "error": traceback.format_exc(), "stdout": out.getvalue()}

    @staticmethod
    def _read_request(conn):
        """Read one newline-terminated JSON request. Returns dict or None."""
        conn.setblocking(True)
        conn.settimeout(CONN_READ_TIMEOUT_S)
        buf = b""
        while not buf.endswith(b"\n"):
            chunk = conn.recv(4096)
            if not chunk:
                break
            buf += chunk
        if not buf.strip():
            return None
        return json.loads(buf.decode("utf-8"))

    @staticmethod
    def _respond(conn, resp):
        conn.sendall((json.dumps(resp) + "\n").encode("utf-8"))


def _start():
    port = int(os.environ.get("TINQS_BLENDER_BRIDGE_PORT", DEFAULT_PORT))

    # Re-running the script (Text Editor) or re-importing (startup) must not
    # leak the previous socket/timer -- the singleton lives on builtins so it
    # survives this module's namespace being rebuilt.
    old = getattr(builtins, SINGLETON_ATTR, None)
    if old is not None:
        try:
            old.stop()
            _log("closed previous bridge instance before restart")
        except Exception:
            pass
        setattr(builtins, SINGLETON_ATTR, None)

    bridge = _Bridge(port)
    try:
        bridge.start()
    except OSError:
        _log("BIND FAILED on {}:{}:\n{}".format(HOST, port, traceback.format_exc()))
        print("TinqsBlenderBridge: port {} busy -- another Blender serving? see {}".format(
            port, LOG_PATH))
        return
    setattr(builtins, SINGLETON_ATTR, bridge)


def _deferred_start():
    """One-shot timer body: runs after startup, when the full API is live."""
    try:
        _start()
    except Exception:
        _log("deferred start CRASHED:\n{}".format(traceback.format_exc()))
    return None


def register():
    """Called by Blender for modules in scripts/startup/ at launch.

    Much of bpy is restricted during startup registration (bpy.data is a
    _RestrictData stub), so we only schedule the real start here; the timer
    fires once Blender is fully up.
    """
    bpy.app.timers.register(_deferred_start, first_interval=0.1)


def unregister():
    bridge = getattr(builtins, SINGLETON_ATTR, None)
    if bridge is not None:
        bridge.stop()
        setattr(builtins, SINGLETON_ATTR, None)


if __name__ == "__main__":
    _start()