animation/tools/md_bridge.py

#!/usr/bin/env python
"""md_bridge.py -- terminal client for the TinqsMDBridge Marvelous Designer plug-in.

Talks newline-delimited JSON over a localhost TCP socket to the TinqsMDBridge
plug-in (tools/md_bridge/TinqsMDBridge.py, registered manually via MD's
Plugin tab > Plug-in Manager > + ADD). Clicking Plugin > TinqsMDBridge starts
a BRIDGE SESSION: MD's UI freezes and the plug-in serves requests on the main
thread until --stop is sent or the idle timeout (default 300 s) expires. MD
cannot be used interactively during a session -- that's by design, see
docs/md-bridge.md.

Same wire protocol and `result` convention as tools/iclone_bridge.py, but on
port 18900 so both bridges can run at once.

Usage:
  python tools/md_bridge.py --ping
  python tools/md_bridge.py --exec "result = dir(pattern_api)"
  python tools/md_bridge.py --file some_script.py
  python tools/md_bridge.py --stop
  python tools/md_bridge.py --port 18901 --timeout 120 --exec "..."

The exec namespace is persistent per MD session and pre-seeded with MD's api
modules (import_api, export_api, pattern_api, fabric_api, utility_api,
ApiTypes) plus a BRIDGE info dict. See docs/md-bridge.md.
"""

import argparse
import json
import socket
import sys
import time

DEFAULT_PORT = 18900
DEFAULT_TIMEOUT = 30.0

# Sent by --ping: bridge metadata seeded by the plug-in at startup.
PING_CODE = "result = BRIDGE\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 MD 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 Marvelous Designer.

    Returns the parsed response dict. Raises OSError on transport failure;
    the 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 TinqsMDBridge Marvelous Designer socket plug-in.")
    p.add_argument("--port", type=int, default=DEFAULT_PORT,
                   help="bridge TCP port (default: %(default)s; must match "
                        "TINQS_MD_BRIDGE_PORT inside MD if overridden there)")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
                   help="socket timeout in seconds (default: %(default)s; raise "
                        "for long Simulate() calls)")
    group = p.add_mutually_exclusive_group(required=True)
    group.add_argument("--ping", action="store_true",
                       help="check the bridge is alive; reports executor mode, "
                            "api modules, embedded python")
    group.add_argument("--stop", action="store_true",
                       help="end the bridge session (unfreezes MD's UI)")
    group.add_argument("--exec", metavar="CODE",
                       help="python source to exec() inside Marvelous Designer")
    group.add_argument("--file", metavar="PATH",
                       help="path to a python file to exec() inside Marvelous Designer")
    args = p.parse_args(argv)

    try:
        if args.ping:
            resp = ping(port=args.port, timeout=args.timeout)
        elif args.stop:
            resp = run("__STOP__", 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 Marvelous Designer "
            "running? was Plugin > TinqsMDBridge clicked this session?".format(args.port),
            file=sys.stderr,
        )
        return 1
    except OSError as exc:
        print("error: could not reach the MD 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:
            info = resp.get("result") or {}
            py = str(info.get("python", "?")).split()[0]
            print("ok: MD bridge mode={} qt={} python {} api={}".format(
                info.get("mode", "?"), info.get("qt_binding"), py,
                ",".join(info.get("api_modules", []))))
            if info.get("api_missing"):
                print("warning: missing api modules: {}".format(
                    ",".join(info["api_missing"])))
        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())