python - 如何在进程仍在运行时写入进程标准输入并从其标准输出读取?
问题描述
编辑:我在 Windows 上运行它,因为服务器是在我的主要游戏平台上管理的
我正在尝试创建一个 minecraft 服务器控制器,以便我可以简化管理服务器的过程。我已经能够使用 subprocess 毫无问题地启动和停止服务器,但是当我尝试让它发出命令时它不会。除非我使用subprocess.communicate()
此解决方案的问题,否则它会导致脚本永远等待并最终崩溃,从而使服务器运行。这是我所拥有的代码。
import subprocess
class Server:
def __init__(self, server_start_bat, dir):
self.server_start_bat = server_start_bat
self.dir = dir
def start_server(self):
self.server = subprocess.Popen(self.server_start_bat, cwd=self.dir, shell=True, stdin=subprocess.PIPE,
universal_newlines=True, text=True)
def stop_server(self):
self.run_command('stop')
self.server.communicate()
def op_me(self):
self.run_command(f'op Username')
def clear_weather(self):
self.run_command(f'weather clear 1000000')
def mob_griefing_on(self):
self.run_command(f'gamerule mobGriefing True')
def mob_griefing_off(self):
self.run_command(f'gamerule mobGriefing True')
def set_time_day(self):
self.run_command(f'time set day')
def set_time_night(self):
self.run_command(f'time set night')
def run_command(self, command_text):
self.server.stdin.write(f'{command_text}\n')
解决方案
编辑:在 python 中检查 subprocess.PIPE 上的非阻塞读取
我正在与一个非常相似的问题作斗争,我花了 3 天时间才让它全部运行起来。subprocess.communicate 的问题是你只能调用它一次,并且输出似乎只在子进程终止时给出(根据我的经验,很可能是错误的)。
我遇到的问题,恕我直言,您的问题归结为:如何在进程仍在运行时写入标准输入并从其标准输出中读取?
我最终使用了管道。请注意,如果您编写的内容小于 BUFSIZE(在我的情况下为 0x1000),它们必须让我冲洗。有关man pipe
更多信息,请参阅。
无论如何,下面是我的代码。
import time
import pty
import os
import subprocess
PIPE_BUFSIZE = 4096
_control = False
class Pipe:
def __init__(self, flags=None, terminal=True):
"""Creates a Pipe you can easily write to and read from. Default is to open up a pseudo-terminal.
If you supply flags, pipe2 is used."""
if flags or not terminal:
self._readfd, self._writefd = os.pipe2(flags)
else: # default
self._readfd, self._writefd = pty.openpty()
self.readobj = open(self._readfd, "rb", 0)
self.writeobj = open(self._writefd, "wb", 0)
def write(self, text):
if isinstance(text, str):
text = text.encode()
result = self.writeobj.write(text)
self.writeobj.flush()
return result
def read(self, n):
if _control: # im not using this anymore since the flush seems to be reliable
controlstr = b"this_was_flushed"
controllen = len(controlstr)
self.writeobj.write(controlstr)
time.sleep(.001)
self.writeobj.flush()
result = self.readobj.read(n+controllen)
assert result[-controllen:] == controlstr
return result[:-controllen]
else:
self.writeobj.flush()
return self.readobj.read(n)
class ProcessIOWrapper:
"""Provides an easy way to redirect stdout and stderr using pipes. Write to the processes STDIN and read from STDOUT at any time! """
def __init__(self, args, inittext=None, redirect=True):
#create three pseudo terminals
self.in_pipe = Pipe()
self.out_pipe = Pipe()
self.err_pipe = Pipe()
# if we want to redirect, tell the subprocess to write to our pipe, else it will print to normal stdout
if redirect:
stdout_arg= self.out_pipe.writeobj
stderr_arg= self.err_pipe.writeobj
else:
stdout_arg=None
stderr_arg= None
self.process = subprocess.Popen(args, stdin=self.in_pipe.readobj, stdout=stdout_arg, stderr=stderr_arg)
def write(self, text):
return self.in_pipe.write(text)
def read(self, n, start=None):
return self.out_pipe.read(n)
我用于测试的小 C 程序(也使用过其他程序)
#include <stdio.h>
int main(){
puts("start\n");
char buf[100]="bufbufbuf";
while(1){
gets(buf);
for (int i=0; i<strlen(buf); i++)
if (i%2==0) buf[i]+=1;
puts(buf);
}
}
推荐阅读
- macos - 尝试在 MAC 上使用 sed 编辑器交换 JSON 元素
- django - 到期后,令牌不起作用,api 抛出“未经授权的错误”
- azure - Azure function-Timer unscheduledInvocationReason
- ios - 您如何修改使用本机摄像头的 cordova 插件,以便尊重 iPhone X 上的安全区域?
- batch-processing - 了解在 DBOW 的单个时期发生了什么
- javascript - 状态更改后子组件不更新
- android - Android 应用程序链接 - 将网站与多个应用程序关联时的消歧对话框问题
- spring-boot - 为什么我在使用 mapstruct 从 DTO 映射到实体时没有得到 id?
- amazon-web-services - 带有 Internet 网关的 VPC 中的 AWS Lambda 函数仍然无法访问 Internet
- javascript - 为什么我会收到“UnhandledPromiseRejectionWarning”?