send2ue插件支持blender4.0以上

# Copyright Epic Games, Inc. All Rights Reserved. import os import json import time import sys import inspect from xmlrpc.client import ProtocolError from http.client import RemoteDisconnected sys.path.append(os.path.dirname(__file__)) import rpc.factory import remote_execution try: import unreal except ModuleNotFoundError: pass REMAP_PAIRS = [] UNREAL_PORT = int(os.environ.get('UNREAL_PORT', 9998)) # use a different remap pairs when inside a container if os.environ.get('TEST_ENVIRONMENT'): UNREAL_PORT = int(os.environ.get('UNREAL_PORT', 8998)) REMAP_PAIRS = [(os.environ.get('HOST_REPO_FOLDER'), os.environ.get('CONTAINER_REPO_FOLDER'))] # this defines a the decorator that makes function run as remote call in unreal remote_unreal_decorator = rpc.factory.remote_call( port=UNREAL_PORT, default_imports=['import unreal'], remap_pairs=REMAP_PAIRS, ) rpc_client = rpc.client.RPCClient(port=UNREAL_PORT) unreal_response = '' def get_response(): """ Gets the stdout produced by the remote python call. :return str: The stdout produced by the remote python command. """ if unreal_response: full_output = [] output = unreal_response.get('output') if output: full_output.append('\n'.join([line['output'] for line in output if line['type'] != 'Warning'])) result = unreal_response.get('result') if result != 'None': full_output.append(result) return '\n'.join(full_output) return '' def add_indent(commands, indent): """ Adds an indent to the list of python commands. :param list commands: A list of python commands that will be run by unreal engine. :param str indent: A str of tab characters. :return str: A list of python commands that will be run by unreal engine. """ indented_line = [] for command in commands: for line in command.split('\n'): indented_line.append(f'{indent}{line}') return indented_line def print_python(commands): """ Prints the list of commands as formatted output for debugging and development. :param list commands: A list of python commands that will be run by unreal engine. """ if os.environ.get('REMOTE_EXECUTION_SHOW_PYTHON'): dashes = '-' * 50 label = 'Remote Execution' sys.stdout.write(f'{dashes}{label}{dashes}\n') # get the function name current_frame = inspect.currentframe() caller_frame = inspect.getouterframes(current_frame, 2) function_name = caller_frame[3][3] kwargs = caller_frame[3][0].f_locals kwargs.pop('commands', None) kwargs.pop('result', None) # write out the function name and its arguments sys.stdout.write( f'{function_name}(kwargs={json.dumps(kwargs, indent=2, default=lambda element: type(element).__name__)})\n') # write out the code with the lines numbers for index, line in enumerate(commands, 1): sys.stdout.write(f'{index} {line}\n') sys.stdout.write(f'{dashes}{"-" * len(label)}{dashes}\n') def run_unreal_python_commands(remote_exec, commands, failed_connection_attempts=0): """ Finds the open unreal editor with remote connection enabled, and sends it python commands. :param object remote_exec: A RemoteExecution instance. :param list commands: A list of python commands that will be run by unreal engine. :param int failed_connection_attempts: A counter that keeps track of how many times an editor connection attempt was made. """ if failed_connection_attempts == 0: print_python(commands) # wait a tenth of a second before attempting to connect time.sleep(0.1) try: # try to connect to an editor for node in remote_exec.remote_nodes: remote_exec.open_command_connection(node.get("node_id")) # if a connection is made if remote_exec.has_command_connection(): # run the import commands and save the response in the global unreal_response variable global unreal_response unreal_response = remote_exec.run_command('\n'.join(commands), unattended=False) # otherwise make an other attempt to connect to the engine else: if failed_connection_attempts < 50: run_unreal_python_commands(remote_exec, commands, failed_connection_attempts + 1) else: remote_exec.stop() raise ConnectionError("Could not find an open Unreal Editor instance!") # catch all errors except: raise ConnectionError("Could not find an open Unreal Editor instance!") # shutdown the connection finally: remote_exec.stop() return get_response() def run_commands(commands): """ Runs a list of python commands and returns the result of the output. :param list commands: A formatted string of python commands that will be run by unreal engine. :return str: The stdout produced by the remote python command. """ # wrap the commands in a try except so that all exceptions can be logged in the output commands = ['try:'] + add_indent(commands, '\t') + ['except Exception as error:', '\tprint(error)'] # start a connection to the engine that lets you send python-commands.md strings remote_exec = remote_execution.RemoteExecution() remote_exec.start() # send over the python code as a string and run it return run_unreal_python_commands(remote_exec, commands) def is_connected(): """ Checks the rpc server connection """ try: return rpc_client.proxy.is_running() except (RemoteDisconnected, ConnectionRefusedError, ProtocolError): return False def set_rpc_env(key, value): """ Sets an env value on the unreal RPC server. """ rpc_client.proxy.set_env(key, value) def bootstrap_unreal_with_rpc_server(): """ Bootstraps the running unreal editor with the unreal rpc server if it doesn't already exist. """ if not os.environ.get('TEST_ENVIRONMENT'): if not is_connected(): import bpy rpc_response_timeout = bpy.context.preferences.addons["send2ue"].preferences.rpc_response_timeout dependencies_path = os.path.dirname(__file__) result = run_commands( [ 'import sys', 'import os', 'import threading', 'for thread in threading.enumerate():', '\tif thread.name =="UnrealRPCServer":', '\t\tthread.kill()', f'os.environ["RPC_TIME_OUT"] = "{rpc_response_timeout}"', f'sys.path.append(r"{dependencies_path}")', 'from rpc import unreal_server', 'rpc_server = unreal_server.RPCServer()', 'rpc_server.start(threaded=True)', ] ) if result: raise ConnectionError(result) class Unreal: @staticmethod def get_value(value, unreal_type=None): """ Gets the value as an unreal type. :param Any value: A value that can be any generic python type. :param str unreal_type: The name of an unreal type. :return Any: The converted unreal value. """ if unreal_type == 'Array': if isinstance(value, str): value = value.split(',') if value: array = unreal.Array(type(value[0])) for element in value: array.append(element) return array elif unreal_type == 'Int32Interval': i