From e6bcfe5f8661cbb7bcd568c1209da3329972884a Mon Sep 17 00:00:00 2001 From: Saifeddine ALOUI Date: Sat, 13 Jul 2024 13:59:46 +0200 Subject: [PATCH] upgraed execution code --- tests/execution/test_execution.py | 30 ++++++ .../python_execution_engine.py | 99 ++++++++++++++++++- 2 files changed, 127 insertions(+), 2 deletions(-) create mode 100644 tests/execution/test_execution.py diff --git a/tests/execution/test_execution.py b/tests/execution/test_execution.py new file mode 100644 index 00000000..afb0649d --- /dev/null +++ b/tests/execution/test_execution.py @@ -0,0 +1,30 @@ +import subprocess +import platform +from pathlib import Path +from ascii_colors import get_trace_exception, trace_exception + +if __name__ == "__main__": + # Create a temporary file. + code = "print('Hello world');input('hi')" + message_id=102 + root_folder = Path(r"E:\lollms\discussion_databases\html stuff\105") + root_folder.mkdir(parents=True, exist_ok=True) + tmp_file = root_folder / f"ai_code_{message_id}.py" + with open(tmp_file, "w", encoding="utf8") as f: + f.write(code) + + try: + # Determine the platform and open a terminal to execute the Python code. + system = platform.system() + if system == "Windows": + subprocess.Popen(f"""start cmd /k "cd /d {root_folder} && python {tmp_file} && pause" """, shell=True) + elif system == "Darwin": # macOS + subprocess.Popen(["open", "-a", "Terminal", f'cd "{root_folder}" && python "{tmp_file}"'], shell=True) + elif system == "Linux": + subprocess.Popen(["x-terminal-emulator", "-e", f'bash -c "cd \\"{root_folder}\\" && python \\"{tmp_file}\\"; exec bash"'], shell=True) + else: + raise Exception(f"Unsupported platform: {system}") + + except Exception as ex: + error_message = f"Error executing Python code: {ex}" + error_json = {"output": "
" + error_message + "\n" + get_trace_exception(ex) + "
", "execution_time": 0} diff --git a/utilities/execution_engines/python_execution_engine.py b/utilities/execution_engines/python_execution_engine.py index 0828fed5..546b6fd2 100644 --- a/utilities/execution_engines/python_execution_engine.py +++ b/utilities/execution_engines/python_execution_engine.py @@ -13,10 +13,71 @@ import time import subprocess import json from lollms.client_session import Client - +import platform +from pathlib import Path lollmsElfServer:LOLLMSWebUI = LOLLMSWebUI.get_instance() -def execute_python(code, client:Client, message_id, build_file=True): + +def execute_python(code, client, message_id, build_file=True): + def spawn_process(code): + """Executes Python code in a new terminal and returns the output as JSON.""" + + # Start the timer. + start_time = time.time() + + # Create a temporary file. + root_folder = client.discussion.discussion_folder + root_folder.mkdir(parents=True, exist_ok=True) + tmp_file = root_folder / f"ai_code_{message_id}.py" + with open(tmp_file, "w", encoding="utf8") as f: + f.write(code) + + try: + # Determine the platform and open a terminal to execute the Python code. + system = platform.system() + if system == "Windows": + process = subprocess.Popen(f"""start cmd /k "cd /d "{root_folder}" && python "{tmp_file}" && pause" """, shell=True) + elif system == "Darwin": # macOS + process = subprocess.Popen(["open", "-a", "Terminal", f'cd "{root_folder}" && python "{tmp_file}"'], shell=True) + elif system == "Linux": + process = subprocess.Popen(["x-terminal-emulator", "-e", f'bash -c "cd \\"{root_folder}\\" && python \\"{tmp_file}\\"; exec bash"'], shell=True) + else: + raise Exception(f"Unsupported platform: {system}") + + # Wait for the process to complete. + process.wait() + + # Get the output and error from the process. + output, error = process.communicate() + except Exception as ex: + # Stop the timer. + execution_time = time.time() - start_time + error_message = f"Error executing Python code: {ex}" + error_json = {"output": "
" + error_message + "\n" + get_trace_exception(ex) + "
", "execution_time": execution_time} + return error_json + + # Stop the timer. + execution_time = time.time() - start_time + + # Check if the process was successful. + if process.returncode != 0: + # The child process threw an exception. + try: + error_message = f"Output: {output.decode('utf-8', errors='ignore')}\nError executing Python code:\n{error.decode('utf-8', errors='ignore')}" + except: + error_message = f"Error executing Python code:\n{error}" + error_json = {"output": "
" + error_message + "
", "execution_time": execution_time} + return error_json + + # The child process was successful. + output_json = {"output": output.decode("utf8"), "execution_time": execution_time} + return output_json + + return spawn_process(code) + + + +def execute_python_old(code, client:Client, message_id, build_file=True): def spawn_process(code): """Executes Python code and returns the output as JSON.""" @@ -65,3 +126,37 @@ def execute_python(code, client:Client, message_id, build_file=True): output_json = {"output": output.decode("utf8"), "execution_time": execution_time} return output_json return spawn_process(code) + + +def create_and_execute_script(code, message_id, root_folder): + try: + # Ensure the root folder exists + root_folder = Path(root_folder) + root_folder.mkdir(parents=True, exist_ok=True) + + # Create the temporary Python file + tmp_file = root_folder / f"ai_code_{message_id}.py" + with open(tmp_file, "w", encoding="utf8") as f: + f.write(code) + + # Determine the platform and open a terminal to execute the Python code + system = platform.system() + if system == "Windows": + subprocess.Popen(f"""start cmd /k "cd /d "{root_folder}" && python "{tmp_file}" && pause" """, shell=True) + elif system == "Darwin": # macOS + subprocess.Popen(["open", "-a", "Terminal", f'cd "{root_folder}" && python "{tmp_file}"'], shell=True) + elif system == "Linux": + subprocess.Popen(["x-terminal-emulator", "-e", f'bash -c "cd \\"{root_folder}\\" && python \\"{tmp_file}\\"; exec bash"'], shell=True) + else: + raise Exception(f"Unsupported platform: {system}") + + except Exception as ex: + error_message = f"Error executing Python code: {ex}" + error_json = {"output": "
" + error_message + "\n" + get_trace_exception(ex) + "
", "execution_time": 0} + print(error_json) + +if __name__ == "__main__": + code = "print('Hello world');input('hi')" + message_id = 102 + root_folder = r"E:\lollms\discussion_databases\html stuff\105" + create_and_execute_script(code, message_id, root_folder) \ No newline at end of file