这甚至可能吗?将命令/对象从一个 python shell 发送到另一个? [英] is this even possible? send commands/objects from one python shell to another?
本文介绍了这甚至可能吗?将命令/对象从一个 python shell 发送到另一个?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
我有一个问题，经过一番挖掘后我无法真正解决，这也不是我的专业领域，所以我什至不知道我在寻找什么.
我想知道是否可以将两个 python shell链接"在一起?
这是实际用例...
我正在使用一个程序，该程序在 GUI 中内置了自己的专用 python shell.当您在内部 python shell 中运行命令时，GUI 会实时更新以反映您运行的命令.
问题是，脚本环境很糟糕.它基本上是外壳旁边的文本板，只是不断复制和粘贴，不可能真正实现认真的开发.
我想要做的是打开我的 IDE(VSCode/Spyder)，这样我就可以有一个合适的环境，但能够在我的 IDE 中运行命令，这些命令以某种方式被发送到软件的内部 python shell.
是否有可能以某种方式检测软件中打开的外壳并在两个 python 实例之间连接/链接或建立管道?所以我可以在两者之间传递命令/python对象，并且每个对象的变量状态基本相同?
最接近我想要的东西是使用 multiprocessing 模块.或者 socket 或 pexpect?
解决方案
这里是拼凑起来的所有部分！
- 多线程使文件处理系统和 Python 交互式 shell 可以同时工作.
- 变量在交互式 shell 和文件之间更新.换句话说，文件和交互式 shell 共享变量、函数、类等.
- 外壳和文件之间的即时更新.
导入线程进口平台导入文本换行导入回溯导入哈希库导入运行导入代码导入时间导入系统导入操作系统def clear_console():"""根据操作系统清除您的控制台."""如果 platform.system() == "Windows":os.system("cls")elif platform.system() in ("Darwin", "Linux"):os.system(清除")def get_file_md5(file_name):""" 获取文件的 md5 哈希值."""使用 open(file_name, "rb") 作为 f:返回 hashlib.md5(f.read()).hexdigest()def track_file(file_name, one_way=False):"""处理外部文件."""# 获取文件的当前 md5.md5 = get_file_md5(file_name)# 第一次运行的标志.first_run = 真# 如果事件被设置，线程通过退出循环优雅地退出.而不是 event_close_thread.is_set():时间.睡眠(0.1)# 获取文件的更新(如果有)md5 哈希值.md5_current = get_file_md5(file_name)如果 md5 != md5_current 或 first_run:md5 = md5_current# 执行文件的内容.尝试:# 收集线程全局作用域来更新主线程的作用域.thread_scope = runpy.run_path(file_name, init_globals=globals())如果不是 one_way:# 用其他线程更新主线程的作用域..全局().更新(线程范围)# 打印仅在第一次运行后更新.如果不是 first_run:print(f'\n{"="*20} 文件 {file_name} 已更新！{"="*20}\n>>>', end="")别的:first_run = 假除了:打印(f'\n{"="*20} 文件 {file_name} 抛出错误！{"="*20}\n {traceback.format_exc()}\n>>>',结束="",)定义轨道(文件名):""" 初始化跟踪线程(必须以 .start() 开头). """print(f'{"="*20} 正在跟踪文件 {file_name}！{"="*20}')返回 threading.Thread(target=track_file, args=(file_name,)).start()如果 __name__ == "__main__":clear_console()# 创建一个线程事件用于垃圾收集和文件锁定.event_close_thread = threading.Event()横幅 = textwrap.dedent(F"""\{"="*20} 进入初始外壳 {"="*20}\n这个 shell 允许共享全局范围Python 文件和 Python 交互式 shell.使用:\t>>>track("script.py", one_way=False)在更新文件script.py"时，此 shell 将执行文件(将 shell 全局变量传递给它)，然后，如果one_way 为 False，将自己的全局变量更新为文件的执行.""")# 开始交互式 shell.code.interact(banner=banner, readfunc=None, local=globals(), exitmsg="")# 优雅地退出线程.event_close_thread.set()# 退出外壳.print(f'\n{"="*20} 退出 Inception Shell {"="*20}')出口()一个班轮:
exec("""\nimport threading\nimport platform\nimport textwrap\nimport traceback\nimport hashlib\nimport runpy\nimport code\nimport time\nimport sys\nimportos\n\n\ndef clear_console():\n \"\"\" 根据操作系统清除你的控制台.\"\"\"\n\n if platform.system() == "Windows":\nos.system("cls")\n elif platform.system() in ("Darwin", "Linux"):\n os.system("clear")\n\n\ndef get_file_md5(file_name):\n\"\"\" 获取文件的 md5 哈希值.\"\"\"\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 \"\"\" 处理外部文件.\"\"\"\n\n # 获取文件的当前 md5.\n md5= get_file_md5(file_name)\n\n # 第一次运行的标志.\n first_run = True\n\n # 如果事件被设置，线程通过退出循环优雅地退出.\n 而不是 event_close_thread.is_set():\n\n time.sleep(0.1)\n\n # 获取更新(如果有的话)) 文件的 md5 哈希.\n md5_current = get_file_md5(file_name)\n 如果 md5 != md5_current 或 first_run:\n md5 = md5_current\n\n # 执行文件的内容.\n try:\n # 收集线程全局作用域更新主线程的作用域.\n thread_scope = runpy.run_path(file_name, init_globals=globals())\n\n if not one_way:\n # 用其他线程更新主线程的作用域..\n globals().update(thread_scope)\n\n # 打印仅在第一次运行后更新.\n 如果不是 first_run:\n print(f'\\n{"="*20} File {file_name} updated!{"="*20}\\n>>>', end="")\n else:\n first_run = False\n\n 除了:\n print(\n f'\\n{"="*20} File {file_name} 抛出错误！{"="*20}\\n {traceback.format_exc()}\\n>>> ',\n end="",\n )\n\n\ndef track(file_name):\n \"\"\" 初始化跟踪线程(必须以 .start() 开头). \"\"\"\n\n print(f'{"="*20} 正在跟踪的文件 {file_name}！{"="*20}')\n return threading.Thread(target=track_file, args=(file_name,)).start()\n\n\nif __name__ == "__main__":\n clear_console()\n\n # 创建用于垃圾收集和文件锁定的线程事件.\n event_close_thread = threading.Event()\n\n banner = textwrap.dedent(\nf\"\"\"\\\n {"="*20} EnteringInception Shell {"="*20}\\n\n 这个 shell 允许在 Python 文件和 Python 交互式 shell 之间共享全局范围.使用:\n\n \\t >>>track("script.py", one_way=False)\n\n 在更新文件 'script.py' 时，此 shell 将执行\n 文件(将 shell 全局变量传递给它)，然后，如果\n one_way 为 False，则将其自己的全局变量更新为\n 文件的执行.\n \"\"\"\n )\n\n # 开始交互式 shell.\n code.interact(banner=banner, readfunc=None, local=globals(), exitmsg="")\n\n # 优雅地退出线程.\n event_close_thread.set()\n\n # 退出 shell.\n print(f'\\n{"="*20} Exiting Inception Shell {"="*20}')\n exit()\n""")<小时>
为您的 Blender shell 尝试以下操作:
导入线程导入回溯导入哈希库导入运行导入时间def get_file_md5(file_name):""" 获取文件的 md5 哈希值."""使用 open(file_name, "rb") 作为 f:返回 hashlib.md5(f.read()).hexdigest()def track_file(file_name, one_way=False):"""处理外部文件."""# 获取文件的当前 md5.md5 = get_file_md5(file_name)# 第一次运行的标志.first_run = 真# 如果事件被设置，线程通过退出循环优雅地退出.而不是 event_close_thread.is_set():时间.睡眠(0.1)# 获取文件的更新(如果有)md5 哈希值.md5_current = get_file_md5(file_name)如果 md5 != md5_current 或 first_run:md5 = md5_current# 执行文件的内容.尝试:# 收集线程全局作用域来更新主线程的作用域.thread_scope = runpy.run_path(file_name, init_globals=globals())如果不是 one_way:# 用其他线程更新主线程的作用域..全局().更新(线程范围)# 打印仅在第一次运行后更新.如果不是 first_run:打印(f'\n{"="*20} 文件 {file_name} 已更新！{"="*20}\n>>>', 结束="")别的:first_run = 假除了:打印(f'\n{"="*20} 文件 {file_name} 抛出错误！{"="*20}\n {traceback.format_exc()}\n>>>',结束="",)定义轨道(文件名):""" 初始化跟踪线程(必须以 .start() 开头). """print(f'{"="*20} 正在跟踪文件 {file_name}！{"="*20}')返回 threading.Thread(target=track_file, args=(file_name,)).start()如果 __name__ == "__main__":# 创建一个线程事件用于垃圾收集和文件锁定.event_close_thread = threading.Event()# 优雅地退出线程.event_close_thread.set()一个班轮:
exec("""\nimport threading\nimport traceback\nimport hashlib\nimport runpy\nimport time\n\n\ndef get_file_md5(file_name):\n \"\"\" 获取文件的 md5 哈希值.\"\"\"\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 \"\"\" 处理外部文件.\"\"\"\n\n # 获取文件的当前 md5.\n md5 =get_file_md5(file_name)\n\n # 第一次运行的标志.\n first_run = True\n\n # 如果事件被设置，线程通过退出循环优雅地退出.\n 而不是 event_close_thread.is_set():\n\n time.sleep(0.1)\n\n # 获取文件的更新(如果有)md5 哈希值.\n md5_current = get_file_md5(file_name)\n 如果 md5 != md5_current 或 first_run:\n md5 = md5_current\n\n # 执行文件内容.\n try:\n # 收集线程全局作用域以更新主线程的作用域.\nthread_scope = runpy.run_path(file_name, init_globals=globals())\n\n 如果不是 one_way:\n # 用其他线程更新主线程的作用域..\n globals().update(thread_scope)\n\n # 打印仅在第一次运行后更新.\n 如果不是 first_run:\n print(\n f'\\n{"="*20} 文件 {file_name} 已更新！{"="*20}\\n>>>', end=""\n )\n else:\n first_run = False\n\n 除了:\n print(\n f'\\n{"="*20} 文件 {file_name} 抛出错误！{"="*20}\\n {traceback.format_exc()}\\n>>> ',\n end="",\n )\n\n\ndef track(file_name):\n \"\"\" 初始化跟踪线程(必须以 .start() 开头).\"\"\"\n\n print(f'{"="*20} 正在跟踪的文件 {file_name}！{"="*20}')\n return threading.Thread(target=track_file, args=(file_name,)).start()\n\n\nif __name__ == "__main__":\n # 为垃圾创建线程事件集合和文件锁.\n event_close_thread = threading.Event()\n\n # 优雅地退出线程.\n event_close_thread.set()\n""")I have a question I wasn't really able to solve after doing a little digging, and this is also not my area of expertise so I really don't even know what I'm looking for.
I'm wondering if it's possible to "link" together two python shells?
This is the actual use case...
I am working with a program that has it's own dedicated python shell built into the GUI. When you run commands in the internal python shell, the GUI updates in real-time reflecting the commands you ran.
The problem is, the scripting environment is terrible. It's basically a textpad next to a shell and is just constant copy and pasting, no serious development could ever really be achieved.
What I want to do is open my IDE (VSCode/Spyder) so I can have a proper environment, but be able to run commands in my IDE that somehow get sent to softwares internal python shell.
Is it possible to somehow detect the open shell in the software and connect/link or make a pipe between the two python instances? So I can pass commands / python objects between the two and basically have the same state of variables in each?
The closest I've come to seeing something like what I want is using the multiprocessing module. Or perhaps socket or pexpect?
Passing data between separately running Python scripts
How to share variables across scripts in python?
Even if it's just one way communication that might work, just want to be able to use this software in a proper development env.
Honestly I don't really have a clue what I'm doing and hoping for some help here..
解决方案
Here is all the pieces put together!
- Multithreading so that both the file processing system and the Python interactive shell can work at the same time.
- Variables are updated between the interactive shell and the file. In other words, the file and the interactive shell shares variables, functions, classes, etc.
- Instant update between shell and file.
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()
One liner:
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""")
Try the following for your 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()
One liner:
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""")
这篇关于这甚至可能吗?将命令/对象从一个 python shell 发送到另一个?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
