Currently if you run an external process from Python the command and locks Blender until it finishes, eg:subprocess.check_output(["sleep", "2"]).
Whats the best way to wrap an external call in way that the user can see whats happening, and can escape from if its taking too long?
External processes can be run while collecting their output using non-blocking file reading.
This example uses a mix-in class (works the same way as Import/ExportHelper) which you can subclass to create your own operator that runs a command.
Each line printed by the external application shows in the header, and pressing escape kills the process.
Notes:
sys.stdout.flush() in Python).ctypes.Example script:
"""
To test this operator, run the script, then open operator search (space-bar)
to execute: External Command Example
"""
import subprocess
import os
# ----------------------------------------------------------------------------
# Portable non-blocking pipe.
#
# This is really black magic on ms-windows!
if os.name == "nt":
def pipe_non_blocking_set(fd):
# Constant could define globally but avoid polluting the name-space
# see: https://stackoverflow.com/a/35052424/432509
import msvcrt
from ctypes import windll, byref, wintypes, WinError, POINTER
from ctypes.wintypes import HANDLE, DWORD, BOOL
LPDWORD = POINTER(DWORD)
PIPE_NOWAIT = wintypes.DWORD(0x00000001)
def pipe_no_wait(pipefd):
SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState
SetNamedPipeHandleState.argtypes = [HANDLE, LPDWORD, LPDWORD, LPDWORD]
SetNamedPipeHandleState.restype = BOOL
h = msvcrt.get_osfhandle(pipefd)
res = windll.kernel32.SetNamedPipeHandleState(h, byref(PIPE_NOWAIT), None, None)
if res == 0:
print(WinError())
return False
return True
return pipe_no_wait(fd)
def pipe_non_blocking_is_error_blocking(ex):
if not isinstance(ex, PortableBlockingIOError):
return False
from ctypes import GetLastError
ERROR_NO_DATA = 232
return (GetLastError() == ERROR_NO_DATA)
PortableBlockingIOError = OSError
else:
def pipe_non_blocking_set(fd):
import fcntl
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
return True
# only to keep compatibility with nt version
def pipe_non_blocking_is_error_blocking(ex):
if not isinstance(ex, PortableBlockingIOError):
return False
return True
PortableBlockingIOError = BlockingIOError
# end magic!
# ----------------------------------------------------------------------------
class SubprocessHelper:
"""
Mix-in class for operators to run commands in a non-blocking way.
This uses a modal operator to manage an external process.
This class defines:
process: The running process object.
Subclass must define:
command: Sequence of arguments to pass to subprocess.Popen
"""
@staticmethod
def _non_blocking_readlines(f, chunk=64):
"""
Iterate over lines, yielding b'' when nothings left
or when new data is not yet available.
"""
fd = f.fileno()
pipe_non_blocking_set(fd)
blocks = []
while True:
try:
data = os.read(fd, chunk)
if not data:
# case were reading finishes with no trailing newline
yield b''.join(blocks)
blocks.clear()
except PortableBlockingIOError as ex:
if not pipe_non_blocking_is_error_blocking(ex):
raise ex
yield b''
continue
while True:
n = data.find(b'\n')
if n == -1:
break
yield b''.join(blocks) + data[:n + 1]
data = data[n + 1:]
blocks.clear()
blocks.append(data)
def _report_output(self):
stdout_line_iter, stderr_line_iter = self._buffer_iter
for line_iter, report_type in (
(stdout_line_iter, {'INFO'}),
(stderr_line_iter, {'WARNING'})
):
while True:
line = next(line_iter).rstrip() # rstrip all, to include \r on windows
if not line:
break
self.report(report_type, line.decode(encoding='utf-8', errors='surrogateescape'))
def _wm_enter(self, context):
wm = context.window_manager
window = context.window
self._timer = wm.event_timer_add(0.1, context.window)
context.window.cursor_set('WAIT')
def _wm_exit(self, context):
wm = context.window_manager
window = context.window
wm.event_timer_remove(self._timer)
window.cursor_set('DEFAULT')
def modal(self, context, event):
wm = context.window_manager
p = self._process
if event.type == 'ESC':
self.cancel(context)
self.report({'INFO'}, "Operation aborted by user.")
return {'CANCELLED'}
elif event.type == 'TIMER':
if p.poll() is not None:
self._report_output()
self._wm_exit(context)
return {'FINISHED'}
self._report_output()
return {'PASS_THROUGH'}
def execute(self, context):
import subprocess
try:
p = subprocess.Popen(
self.command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
except FileNotFoundError as ex:
# Command not found
self.report({'ERROR'}, str(ex))
return {'CANCELLED'}
self._process = p
self._buffer_iter = (
iter(self._non_blocking_readlines(p.stdout)),
iter(self._non_blocking_readlines(p.stderr)),
)
wm = context.window_manager
wm.modal_handler_add(self)
self._wm_enter(context)
return {'RUNNING_MODAL'}
def cancel(self, context):
self._wm_exit(context)
self._process.kill()
# ------------------------------------------------------------------------
# End generic code, now to show an example operator using SubprocessHelper
import bpy
from bpy.types import (
Operator,
)
class MyCommandOperator(Operator, SubprocessHelper):
bl_idname = "wm.my_command"
bl_label = "External Command Example"
command = (
# Run Blender's Python, but could be any other command.
bpy.app.binary_path_python,
# pass in a script to execute
"-c", """
import sys
import time
# Some dummy status reports
status_report = ('Hello World', 'Blender is working!', '... almost done', 'Task Complete!')
for word in status_report:
# Will show in the info area.
print(word)
# flush to ensure we get the output immediately.
sys.stdout.flush()
# wait a second.
time.sleep(1.0)
"""
)
def register():
bpy.utils.register_class(MyCommandOperator)
def unregister():
bpy.utils.unregister_class(MyCommandOperator)
if __name__ == "__main__":
register()
# bpy.ops.wm.my_command('INVOKE_DEFAULT')
