animation/tools/iclone_bridge.py

#!/usr/bin/env python
"""iclone_bridge.py -- terminal client for the TinqsBridge iClone 8 plugin.

Talks newline-delimited JSON over a localhost TCP socket to the TinqsBridge
plugin (tools/iclone_bridge/TinqsBridge/main.py, installed into iClone via
tools/install_iclone_bridge.ps1). iClone must be running with the plugin
loaded (i.e. restarted at least once since install) for any of this to work.

Python 3.12, stdlib only -- no third-party packages.

Usage:
  python tools/iclone_bridge.py --ping
  python tools/iclone_bridge.py --exec "result = 1 + 1"
  python tools/iclone_bridge.py --file some_script.py
  python tools/iclone_bridge.py --port 18801 --timeout 120 --exec "..."

See docs/iclone-bridge.md for the wire protocol, the `result` convention, and
the persistent-namespace / threading rules.
"""

import argparse
import json
import socket
import sys
import time

DEFAULT_PORT = 18800
DEFAULT_TIMEOUT = 30.0

# Sent by --ping. Finds product/version via RApplication, embedded Python via
# sys.version -- see RLPy.py (RApplication.GetProductName/GetProductVersion).
PING_CODE = (
    "import sys, RLPy\n"
    "result = {\n"
    "    'product': RLPy.RApplication.GetProductName(),\n"
    "    'version': RLPy.RApplication.GetProductVersion(),\n"
    "    'python': sys.version,\n"
    "}\n"
)


def _send(sock, obj):
    sock.sendall((json.dumps(obj) + "\n").encode("utf-8"))


def _recv_line(sock, timeout):
    sock.settimeout(timeout)
    buf = b""
    while not buf.endswith(b"\n"):
        chunk = sock.recv(4096)
        if not chunk:
            if buf:
                break
            raise ConnectionError(
                "connection closed by the iClone bridge before a response was received")
        buf += chunk
    return buf.decode("utf-8")


def run(code, port=DEFAULT_PORT, timeout=DEFAULT_TIMEOUT):
    """Send `code` to be exec()'d inside iClone. Returns the parsed response dict.

    Raises OSError (e.g. ConnectionRefusedError, socket.timeout) on transport
    failure; the returned dict's "ok" key reports exec-level success/failure.
    """
    req = {"id": int(time.time() * 1000), "code": code}
    with socket.create_connection(("127.0.0.1", port), timeout=timeout) as sock:
        _send(sock, req)
        line = _recv_line(sock, timeout)
    return json.loads(line)


def ping(port=DEFAULT_PORT, timeout=DEFAULT_TIMEOUT):
    """Convenience wrapper: run(PING_CODE, ...)."""
    return run(PING_CODE, port=port, timeout=timeout)


def _read_code(args):
    if args.exec is not None:
        return args.exec
    with open(args.file, "r", encoding="utf-8") as f:
        return f.read()


def _main(argv=None):
    p = argparse.ArgumentParser(
        description="Client for the TinqsBridge iClone 8 socket plugin.")
    p.add_argument("--port", type=int, default=DEFAULT_PORT,
                    help="bridge TCP port (default: %(default)s; must match "
                         "TINQS_ICLONE_BRIDGE_PORT inside iClone if overridden there)")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
                    help="socket timeout in seconds (default: %(default)s)")
    group = p.add_mutually_exclusive_group(required=True)
    group.add_argument("--ping", action="store_true",
                        help="check the bridge is alive; reports product/version/embedded python")
    group.add_argument("--exec", metavar="CODE",
                        help="python source to exec() inside iClone")
    group.add_argument("--file", metavar="PATH",
                        help="path to a python file to exec() inside iClone")
    args = p.parse_args(argv)

    try:
        if args.ping:
            resp = ping(port=args.port, timeout=args.timeout)
        else:
            resp = run(_read_code(args), port=args.port, timeout=args.timeout)
    except ConnectionRefusedError:
        print(
            "error: connection refused on 127.0.0.1:{} -- is iClone running? "
            "was it restarted after installing the TinqsBridge plugin?".format(args.port),
            file=sys.stderr,
        )
        return 1
    except OSError as exc:
        print("error: could not reach the iClone bridge on 127.0.0.1:{}: {}".format(
            args.port, exc), file=sys.stderr)
        return 1

    if resp.get("stdout"):
        text = resp["stdout"]
        print(text, end="" if text.endswith("\n") else "\n")

    if resp.get("ok"):
        if args.ping:
            result = resp.get("result") or {}
            py_ver = str(result.get("python", "?")).split()[0]
            print("ok: {} {} (embedded python {})".format(
                result.get("product", "?"), result.get("version", "?"), py_ver))
        else:
            print(json.dumps(resp.get("result"), indent=2))
        return 0

    print("error: {}".format(resp.get("error", "unknown error")), file=sys.stderr)
    return 1


if __name__ == "__main__":
    sys.exit(_main())