Hacky blender basilisp nrepl

blender_nrepl.py

# by Simo / maqq
# tl;dr; don't use this but see https://github.com/ikappaki/basilisp-blender

import bpy
import uuid
import traceback
from basilisp import main as basilisp, cli
from basilisp.lang import compiler, runtime
from collections import OrderedDict
import socket
import selectors
import bencodepy

#import site
#import pip

#pip.main(['install', 'basilisp', '--target', site.getsitepackages()[0]])
#pip.main(['install', 'basilisp', '--target', site.getsitepackages()[0]])

opts = compiler.compiler_opts()
basilisp.init(opts)
ctx = compiler.CompilerContext(filename="blender", opts=opts)
ns = runtime.Namespace.get_or_create(runtime.CORE_NS_SYM)
ns_var = runtime.set_current_ns("blender-user")
eof = object()

def eval_str(code):
    return cli.eval_str(code, ctx, ns_var.value, eof)

def eval_editor(text):
    code = bpy.data.texts[text].as_string()
    return eval_str(code)

def eval_file(filepath):
    return cli.eval_file(filepath, ctx, ns_var.value)

def convert(e):
    if isinstance(e, (dict, OrderedDict)):
        return { k.decode(): convert(v.decode() if hasattr(v, "decode") else v) for k, v in e.items() }
    if isinstance(e, list):
        return [convert(v) for v in e]
    return e

class NREPL:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.sel = selectors.DefaultSelector()
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind((self.host, self.port))
        self.sock.setblocking(False)
        self.sock.listen(100)
        self.sel.register(self.sock, selectors.EVENT_READ, self.accept_connection)
        print("Listening connections on", (self.host, self.port))

    def socket_close(self):
        self.sel.close()
        self.sock.shutdown(1)
        self.sock.close()

    def add_session(self, response, message):
        if message.get("session"):
            response["session"] = message.get("session")

    def op_eval(self, message):
        response = { "status": ["done"], "id": message.get("id") }
        try:
            response["value"] = str(eval_str(message.get("code")))
        except Exception as ex:
            print("Exception when evaluating:", ex)
            response["error"] = str(ex)
        response["ns"] = "blender-user"
        self.add_session(response, message)
        return response

    def op_clone(self, message):
        response = { "status": ["started"], "id": message.get("id") }
        response["new-session"] = str(uuid.uuid4())
        return response

    def op_describe(self, message):
        response = { "status": ["done"], "id": message.get("id") }
        ops_list = ["clone", "describe", "eval", "complete", "info", "load-file"]
        response["ops"] = { k: {} for k in ops_list }
        self.add_session(response, message)
        return response

    def op_info(self, message):
        response = { "status": ["done"], "id": message.get("id") }
        return response

    def socket_read(self, connection, mask):
        try:
            data = connection.recv(10000)
            if data:
                message = convert(bencodepy.decode(data))
                op = message.get("op")
                if hasattr(self, "op_" + op):
                    response = getattr(self, "op_" + op)(message)
                    if response is not None:
                        connection.sendall(bencodepy.encode(response))
                else:
                    print("Unknown operation:", op)
            else:
                raise Exception("No data")
        except Exception as ex:
            print("Exception:", ex)
            print(traceback.format_exc())
            self.sel.unregister(connection)
            connection.close()

    def accept_connection(self, sock, mask):
        connection, address = sock.accept()
        print("Accepted connection from", address)
        connection.setblocking(False)
        self.sel.register(connection, selectors.EVENT_READ, self.socket_read)

    def update(self):
        events = self.sel.select(0)
        for key, mask in events:
            callback = key.data
            callback(key.fileobj, mask)

nrepl = NREPL("127.0.0.1", 8889)

def on_frame_change(scene):
    nrepl.update()

eval_str("(ns blender-user (:require clojure.core))")
bpy.app.handlers.frame_change_pre.append(on_frame_change)
bpy.ops.screen.animation_play()