python - 这有可能吗？从一个python shell发送命令/对象到另一个？

Question

我有一个问题，我做了一点挖掘后仍然无法解决，这也不是我的专业领域，所以我什至不知道自己在寻找什么。

我想知道是否可以将两个python shell “链接”在一起？

这是实际的用例...

我正在使用一个程序，该程序在GUI中内置了自己的专用python shell。当您在内部python shell中运行命令时，GUI会实时更新以反射(reflect)您运行的命令。

问题是，脚本环境很糟糕。它基本上是 shell 旁边的一个文本板，只是不断地复制和粘贴，无法真正实现认真的开发。

我想做的是打开我的IDE(VSCode/Spyder)，这样我可以有一个适当的环境，但是能够在IDE中运行以某种方式发送到软件内部python shell的命令。

是否有可能以某种方式检测软件中的打开 shell 并在两个python实例之间建立连接/链接或建立管道？因此，我可以在两者之间传递命令/python对象，并且基本上每个变量都具有相同的变量状态？

我最近看到的就是想要使用multiprocessing模块的东西。还是socket或pexpect？

Passing data between separately running Python scripts

How to share variables across scripts in python?

即使只是一种可能有效的通信方式，也只是希望能够在适当的开发环境中使用此软件。

老实说，我真的不知道我在做什么，并希望在这里有所帮助。

Answer 1

这是所有拼在一起的东西!

多线程，以便文件处理系统和Python交互式Shell可以同时工作。

在交互式 shell 程序和文件之间更新变量。换句话说，文件和交互式 shell 共享变量，函数，类等。

Shell和文件之间的即时更新。

import threading
import platform
import textwrap
import traceback
import hashlib
import runpy
import code
import time
import sys
import os


def clear_console():
    """ Clear your console depending on OS. """

    if platform.system() == "Windows":
        os.system("cls")
    elif platform.system() in ("Darwin", "Linux"):
        os.system("clear")


def get_file_md5(file_name):
    """ Grabs the md5 hash of the file. """

    with open(file_name, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()


def track_file(file_name, one_way=False):
    """ Process external file. """

    # Grabs current md5 of file.
    md5 = get_file_md5(file_name)

    # Flag for the first run.
    first_run = True

    # If the event is set, thread gracefully quits by exiting loop.
    while not event_close_thread.is_set():

        time.sleep(0.1)

        # Gets updated (if any) md5 hash of file.
        md5_current = get_file_md5(file_name)
        if md5 != md5_current or first_run:
            md5 = md5_current

            # Executes the content of the file.
            try:
                # Gather the threads global scope to update the main thread's scope.
                thread_scope = runpy.run_path(file_name, init_globals=globals())

                if not one_way:
                    # Updates main thread's scope with other thread..
                    globals().update(thread_scope)

                # Prints updated only after first run.
                if not first_run:
                    print(f'\n{"="*20} File {file_name} updated! {"="*20}\n>>> ', end="")
                else:
                    first_run = False

            except:
                print(
                    f'\n{"="*20} File {file_name} threw error! {"="*20}\n {traceback.format_exc()}\n>>> ',
                    end="",
                )


def track(file_name):
    """ Initializes tracking thread (must be started with .start()). """

    print(f'{"="*20} File {file_name} being tracked! {"="*20}')
    return threading.Thread(target=track_file, args=(file_name,)).start()


if __name__ == "__main__":
    clear_console()

    # Creates a thread event for garbage collection, and file lock.
    event_close_thread = threading.Event()

    banner = textwrap.dedent(
        f"""\
        {"="*20} Entering Inception Shell {"="*20}\n
        This shell allows the sharing of the global scope between
        Python files and the Python interactive shell. To use:

        \t >>> track("script.py", one_way=False)

        On update of the file 'script.py' this shell will execute the
        file (passing the shells global variables to it), and then, if
        one_way is False, update its own global variables to that of the
        file's execution.
        """
    )

    # Begins interactive shell.
    code.interact(banner=banner, readfunc=None, local=globals(), exitmsg="")

    # Gracefully exits the thread.
    event_close_thread.set()

    # Exits shell.
    print(f'\n{"="*20} Exiting Inception Shell {"="*20}')
    exit()

一类轮:

exec("""\nimport threading\nimport platform\nimport textwrap\nimport traceback\nimport hashlib\nimport runpy\nimport code\nimport time\nimport sys\nimport os\n\n\ndef clear_console():\n    \"\"\" Clear your console depending on OS. \"\"\"\n\n    if platform.system() == "Windows":\n        os.system("cls")\n    elif platform.system() in ("Darwin", "Linux"):\n        os.system("clear")\n\n\ndef get_file_md5(file_name):\n    \"\"\" Grabs the md5 hash of the file. \"\"\"\n\n    with open(file_name, "rb") as f:\n        return hashlib.md5(f.read()).hexdigest()\n\n\ndef track_file(file_name, one_way=False):\n    \"\"\" Process external file. \"\"\"\n\n    # Grabs current md5 of file.\n    md5 = get_file_md5(file_name)\n\n    # Flag for the first run.\n    first_run = True\n\n    # If the event is set, thread gracefully quits by exiting loop.\n    while not event_close_thread.is_set():\n\n        time.sleep(0.1)\n\n        # Gets updated (if any) md5 hash of file.\n        md5_current = get_file_md5(file_name)\n        if md5 != md5_current or first_run:\n            md5 = md5_current\n\n            # Executes the content of the file.\n            try:\n                # Gather the threads global scope to update the main thread's scope.\n                thread_scope = runpy.run_path(file_name, init_globals=globals())\n\n                if not one_way:\n                    # Updates main thread's scope with other thread..\n                    globals().update(thread_scope)\n\n                # Prints updated only after first run.\n                if not first_run:\n                    print(f'\\n{"="*20} File {file_name} updated! {"="*20}\\n>>> ', end="")\n                else:\n                    first_run = False\n\n            except:\n                print(\n                    f'\\n{"="*20} File {file_name} threw error! {"="*20}\\n {traceback.format_exc()}\\n>>> ',\n                    end="",\n                )\n\n\ndef track(file_name):\n    \"\"\" Initializes tracking thread (must be started with .start()). \"\"\"\n\n    print(f'{"="*20} File {file_name} being tracked! {"="*20}')\n    return threading.Thread(target=track_file, args=(file_name,)).start()\n\n\nif __name__ == "__main__":\n    clear_console()\n\n    # Creates a thread event for garbage collection, and file lock.\n    event_close_thread = threading.Event()\n\n    banner = textwrap.dedent(\n        f\"\"\"\\\n        {"="*20} Entering Inception Shell {"="*20}\\n\n        This shell allows the sharing of the global scope between\n        Python files and the Python interactive shell. To use:\n\n        \\t >>> track("script.py", one_way=False)\n\n        On update of the file 'script.py' this shell will execute the\n        file (passing the shells global variables to it), and then, if\n        one_way is False, update its own global variables to that of the\n        file's execution.\n        \"\"\"\n    )\n\n    # Begins interactive shell.\n    code.interact(banner=banner, readfunc=None, local=globals(), exitmsg="")\n\n    # Gracefully exits the thread.\n    event_close_thread.set()\n\n    # Exits shell.\n    print(f'\\n{"="*20} Exiting Inception Shell {"="*20}')\n    exit()\n""")

为您的Blender shell尝试以下操作:

import threading
import traceback
import hashlib
import runpy
import time


def get_file_md5(file_name):
    """ Grabs the md5 hash of the file. """

    with open(file_name, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()


def track_file(file_name, one_way=False):
    """ Process external file. """

    # Grabs current md5 of file.
    md5 = get_file_md5(file_name)

    # Flag for the first run.
    first_run = True

    # If the event is set, thread gracefully quits by exiting loop.
    while not event_close_thread.is_set():

        time.sleep(0.1)

        # Gets updated (if any) md5 hash of file.
        md5_current = get_file_md5(file_name)
        if md5 != md5_current or first_run:
            md5 = md5_current

            # Executes the content of the file.
            try:
                # Gather the threads global scope to update the main thread's scope.
                thread_scope = runpy.run_path(file_name, init_globals=globals())

                if not one_way:
                    # Updates main thread's scope with other thread..
                    globals().update(thread_scope)

                # Prints updated only after first run.
                if not first_run:
                    print(
                        f'\n{"="*20} File {file_name} updated! {"="*20}\n>>> ', end=""
                    )
                else:
                    first_run = False

            except:
                print(
                    f'\n{"="*20} File {file_name} threw error! {"="*20}\n {traceback.format_exc()}\n>>> ',
                    end="",
                )


def track(file_name):
    """ Initializes tracking thread (must be started with .start()). """

    print(f'{"="*20} File {file_name} being tracked! {"="*20}')
    return threading.Thread(target=track_file, args=(file_name,)).start()


if __name__ == "__main__":
    # Creates a thread event for garbage collection, and file lock.
    event_close_thread = threading.Event()

    # Gracefully exits the thread.
    event_close_thread.set()

一类轮:

exec("""\nimport threading\nimport traceback\nimport hashlib\nimport runpy\nimport time\n\n\ndef get_file_md5(file_name):\n    \"\"\" Grabs the md5 hash of the file. \"\"\"\n\n    with open(file_name, "rb") as f:\n        return hashlib.md5(f.read()).hexdigest()\n\n\ndef track_file(file_name, one_way=False):\n    \"\"\" Process external file. \"\"\"\n\n    # Grabs current md5 of file.\n    md5 = get_file_md5(file_name)\n\n    # Flag for the first run.\n    first_run = True\n\n    # If the event is set, thread gracefully quits by exiting loop.\n    while not event_close_thread.is_set():\n\n        time.sleep(0.1)\n\n        # Gets updated (if any) md5 hash of file.\n        md5_current = get_file_md5(file_name)\n        if md5 != md5_current or first_run:\n            md5 = md5_current\n\n            # Executes the content of the file.\n            try:\n                # Gather the threads global scope to update the main thread's scope.\n                thread_scope = runpy.run_path(file_name, init_globals=globals())\n\n                if not one_way:\n                    # Updates main thread's scope with other thread..\n                    globals().update(thread_scope)\n\n                # Prints updated only after first run.\n                if not first_run:\n                    print(\n                        f'\\n{"="*20} File {file_name} updated! {"="*20}\\n>>> ', end=""\n                    )\n                else:\n                    first_run = False\n\n            except:\n                print(\n                    f'\\n{"="*20} File {file_name} threw error! {"="*20}\\n {traceback.format_exc()}\\n>>> ',\n                    end="",\n                )\n\n\ndef track(file_name):\n    \"\"\" Initializes tracking thread (must be started with .start()). \"\"\"\n\n    print(f'{"="*20} File {file_name} being tracked! {"="*20}')\n    return threading.Thread(target=track_file, args=(file_name,)).start()\n\n\nif __name__ == "__main__":\n    # Creates a thread event for garbage collection, and file lock.\n    event_close_thread = threading.Event()\n\n    # Gracefully exits the thread.\n    event_close_thread.set()\n""")