I want to use the subprocess.Popen method to execute the python file and check the input and output stream to complete the interaction, but if the file contains the input function, the original reason is to output the input parameter first, and then get the input, but my code turns out to have to get the input first. Only then can the input parameters be detected from the input stream。
from subprocess import Popen, PIPE
from threading import Thread, Event
from queue import Queue, Empty
class Actuator:
"""一个基本的代码执行器"""
def __init__(self, script_path, receiver_widget):
# 执行文件目录
self.script_path = script_path
# 接受返回结果控件id
self.receiver_widget = receiver_widget
# 线程锁
self.thread_lock = False
# 占用线程
self.thread_current = None
# 用户输入的列表
self.user_input = self.receiver_widget.receiver.receiver_input_list
# 用于线程间通信的队列
self.input_queue = Queue()
def _receiver(self, data: str):
Thread(target=self.receiver_widget.insert, args=('end', data), daemon=True).start()
def _receiver_input(self, *args):
data = self.receiver_widget.get_input()
return data
def run(self):
process = Popen(
# 启动无缓冲模式
["python", "-u", self.script_path],
stdin=PIPE,
stdout=PIPE,
stderr=PIPE,
text=True,
bufsize=1, # 行缓冲
universal_newlines=True,
encoding="utf-8"
)
def read_stream(stream, stream_type):
"""读取子进程的输出流（stdout 或 stderr）"""
# 如果子进程任在运行
while process.poll() is None:
output = stream.read()
# output = stream.readline()
if output:
self._receiver(f"{output.strip()}")
else:
break # 流已关闭
# 启动输出读取线程
stdout_thread = Thread(target=read_stream, args=(process.stdout, "stdout"), daemon=True)
# stderr_thread = Thread(target=read_stream, args=(process.stderr, "stderr"), daemon=True)
stdout_thread.start()
# stderr_thread.start()
# 主线程处理输入
while process.poll() is None:
try:
user_input = self.input_queue.get(timeout=0.1)
process.stdin.write(user_input + "\n")
process.stdin.flush()
# 重置等待状态
except Empty:
if len(self.user_input) == 0:
continue
else:
self.send_input(self.user_input[0])
del self.user_input[0]
# 等待输出线程结束
stdout_thread.join()
# stderr_thread.join()
# 确保子进程终止
if process.poll() is None:
process.terminate()
process.wait()
def send_input(self, user_input: str):
# 将用户输入压住队列
self.input_queue.put(user_input)
Dear friend, please help this poor old man