blender-package-manager-addon/subprocess_adapter.py

import logging
from multiprocessing import Process, Pipe
from bpy.types import Operator

def subprocess_operator(cls: Operator, polling_interval=.01) -> Operator:
    """
    Class decorator which wraps Operator methods with setup code for running a subprocess.

    Expects args for Process() to defined in cls.proc_args and cls.proc_kwargs. For example,
    setting cls.proc_kwargs = {'target:' some_func} can be used to run 'some_func' in
    a subprocess.
    """

    def decoratify(cls, methodname, decorator):
        """
        Decorate `cls.methodname` with `decorator` if method `methodname` exists in cls
        else just call the decorator with an empty function
        """
        orig_method = getattr(cls, methodname, None)
        if orig_method:
            setattr(cls, methodname, decorator(orig_method))
        else:
            # HACK: need a no-op which accepts any arguments
            setattr(cls, methodname, decorator(lambda *args, **kwargs: None))


    def decorate_execute(orig_execute):
        def execute(self, context):
            orig_execute(self, context)
            call_copy_of_method_if_exist(cls, 'execute', self, context)
            return self.invoke(context, None)
        return execute
    

    def decorate_invoke(orig_invoke):
        def invoke(self, context, event):
            orig_invoke(self, context, event)

            self.pipe = Pipe()
            self.proc = Process(
                    *getattr(self, 'proc_args', []),
                    **getattr(self, 'proc_kwargs', [])
                    )
            self.proc.start()
            # we have to explicitly close the end of the pipe we are NOT using,
            # otherwise no exception will be generated when the other process closes its end.
            self.pipe[1].close()

            wm = context.window_manager
            wm.modal_handler_add(self)
            self.timer = wm.event_timer_add(polling_interval, context.window)

            return {'RUNNING_MODAL'}
        return invoke

    def poll_subprocess(self):
        self.log.debug("polling")
        try:
            if self.pipe[0].poll():
                newdata = self.pipe[0].recv()
            else:
                newdata = None
        except EOFError:
            self.log.debug("done polling")
            return {'FINISHED'}

        if newdata is not None:
            self.handle_response(newdata) #TODO: make this a customizable callback
            # this should allow chaining of multiple subprocess in a single operator

        return {'PASS_THROUGH'}

    decoratify(cls, 'execute', decorate_execute)
    decoratify(cls, 'invoke', decorate_invoke)
    setattr(cls, 'poll_subprocess', poll_subprocess)

    return cls