代码之家 › 专栏 › 技术社区 › Jonathan Root Josep Valls

具有实时输入和多个控制台的Python子流程

real-time subprocess console input python

Jonathan Root Josep Valls · 技术社区 · 7 年前

问题是,我似乎能够发送到第二个控制台的stdin,但是,在第二个控制台上没有打印任何内容。

writing(input, pipe, process) 包含将生成的字符串复制到as的部件 pipe 通过子流程打开的控制台传递stdin。函数写入(…)是通过类运行的 writetest(Thread)

import os
import sys
import io
import time
import threading
from cmd import Cmd
from queue import Queue
from subprocess import Popen, PIPE, CREATE_NEW_CONSOLE


REPETITIONS = 3


# Position of "The class" (Edit-2)


# Position of "The class" (Edit-1)


class generatetest(threading.Thread):

    def __init__(self, queue):
        self.output = queue
        threading.Thread.__init__(self)

    def run(self):
        print('run generatetest')
        generating(REPETITIONS, self.output)
        print('generatetest done')

    def getout(self):
        return self.output


class writetest(threading.Thread):

    def __init__(self, input=None, pipe=None, process=None):
        if (input == None):        # just in case
            self.input = Queue()
        else:
            self.input = input

        if (pipe == None):        # just in case
            self.pipe = PIPE
        else:
            self.pipe = pipe

        if (process == None):        # just in case
            self.process = subprocess.Popen('C:\Windows\System32\cmd.exe', universal_newlines=True, creationflags=CREATE_NEW_CONSOLE)
        else:
            self.process = proc

        threading.Thread.__init__(self)

    def run(self):
        print('run writetest')
        writing(self.input, self.pipe, self.process)
        print('writetest done')


# Position of "The function" (Edit-2)


# Position of "The function" (Edit-1)


def generating(maxint, outline):
    print('def generating')
    for i in range(maxint):
        time.sleep(1)
        outline.put_nowait(i)


def writing(input, pipe, process):
    print('def writing')
    while(True):
        try:
            print('try')
            string = str(input.get(True, REPETITIONS)) + "\n"
            pipe = io.StringIO(string)
            pipe.flush()
            time.sleep(1)
            # print(pipe.readline())
        except:
            print('except')
            break
        finally:
            print('finally')
            pass


data_queue = Queue()
data_pipe = sys.stdin
# printer = sys.stdout
# data_pipe = os.pipe()[1]


# The code of 'C:\\Users\\Public\\Documents\\test\\test-cmd.py'
# can be found in the question's text further below under "More code"


exe = 'C:\Python34\python.exe'
# exe = 'C:\Windows\System32\cmd.exe'
arg = 'C:\\Users\\Public\\Documents\\test\\test-cmd.py'
arguments = [exe, arg]
# proc = Popen(arguments, universal_newlines=True, creationflags=CREATE_NEW_CONSOLE)
proc = Popen(arguments, stdin=data_pipe, stdout=PIPE, stderr=PIPE,
             universal_newlines=True, creationflags=CREATE_NEW_CONSOLE)


# Position of "The call" (Edit-2 & Edit-1) - file init (proxyfile)


# Position of "The call" (Edit-2) - thread = sockettest()
# Position of "The call" (Edit-1) - thread0 = logtest()
thread1 = generatetest(data_queue)
thread2 = writetest(data_queue, data_pipe, proc)
# time.sleep(5)


# Position of "The call" (Edit-2) - thread.start()
# Position of "The call" (Edit-1) - thread0.start()
thread1.start()
thread2.start()


# Position of "The call" (Edit-2) - thread.join()
# Position of "The call" (Edit-1) - thread.join()
thread1.join(REPETITIONS * REPETITIONS)
thread2.join(REPETITIONS * REPETITIONS)

# data_queue.join()
# receiver = proc.communicate(stdin, 5)
# print('OUT:' + receiver[0])
# print('ERR:' + receiver[1])

print("1st part finished")

略为不同的方法

proc2 = Popen(['C:\Python34\python.exe', '-i'],
              stdin=PIPE,
              stdout=PIPE,
              stderr=PIPE,
              creationflags=CREATE_NEW_CONSOLE)
proc2.stdin.write(b'2+2\n')
proc2.stdin.flush()
print(proc2.stdout.readline())
proc2.stdin.write(b'len("foobar")\n')
proc2.stdin.flush()
print(proc2.stdout.readline())
time.sleep(1)
proc2.stdin.close()
proc2.terminate()
proc2.wait(timeout=0.2)

print("Exiting Main Thread")

编辑-1:通过文件解决方案(概念验证)

Working multiple consoles in python ,一般来说是有效的。但是,由于日志文件将增长到许多GB,因此在这种情况下,它不是一个实用的解决方案。它至少需要文件拆分和正确处理。

班级:

class logtest(threading.Thread):

    def __init__(self, file):
        self.file = file
        threading.Thread.__init__(self)

    def run(self):
        print('run logtest')
        logging(self.file)
        print('logtest done')

职能:

def logging(file):
    pexe = 'C:\Python34\python.exe '
    script = 'C:\\Users\\Public\\Documents\\test\\test-004.py'
    filek = '--file'
    filev = file

    file = open(file, 'a')
    file.close()
    time.sleep(1)

    print('LOG START (outer): ' + script + ' ' + filek + ' ' + filev)
    proc = Popen([pexe, script, filek, filev], universal_newlines=True, creationflags=CREATE_NEW_CONSOLE)
    print('LOG FINISH (outer): ' + script + ' ' + filek + ' ' + filev)

    time.sleep(2)

电话:

# The file tempdata is filled with several strings of "0\n1\n2\n"
# Looking like this:
# 0
# 1
# 2
# 0
# 1
# 2

proxyfile = 'C:\\Users\\Public\\Documents\\test\\tempdata'
f = open(proxyfile, 'a')
f.close()
time.sleep(1)

thread0 = logtest(proxyfile)
thread0.start()
thread0.join(REPETITIONS * REPETITIONS)

由于Windows不提供tail命令,因此我使用了以下脚本(基于 How to implement a pythonic equivalent of tail -F? class CommandLine(Cmd) 最初试图保持第二个控制台打开(因为缺少脚本文件参数)。不过,它也证明了它对保持控制台流畅地打印新日志文件内容是有用的。否则,输出不是确定性的/可预测的。

import time
import sys
import os
import threading
from cmd import Cmd
from argparse import ArgumentParser


def main(args):
    parser = ArgumentParser(description="Parse arguments.")
    parser.add_argument("-f", "--file", type=str, default='', required=False)
    arguments = parser.parse_args(args)

    if not arguments.file:
        print('LOG PRE-START (inner): file argument not found. Creating new default entry.')
        arguments.file = 'C:\\Users\\Public\\Documents\\test\\tempdata'

    print('LOG START (inner): ' + os.path.abspath(os.path.dirname(__file__)) + ' ' + arguments.file)

    f = open(arguments.file, 'a')
    f.close()
    time.sleep(1)

    words = ['word']
    console = CommandLine(arguments.file, words)
    console.prompt = ''

    thread = threading.Thread(target=console.cmdloop, args=('', ))
    thread.start()
    print("\n")

    for hit_word, hit_sentence in console.watch():
        print("Found %r in line: %r" % (hit_word, hit_sentence))

    print('LOG FINISH (inner): ' + os.path.abspath(os.path.dirname(__file__)) + ' ' + arguments.file)


class CommandLine(Cmd):
    """Custom console"""

    def __init__(self, fn, words):
        Cmd.__init__(self)
        self.fn = fn
        self.words = words

    def watch(self):
        fp = open(self.fn, 'r')
        while True:
            time.sleep(0.05)
            new = fp.readline()
            print(new)
            # Once all lines are read this just returns ''
            # until the file changes and a new line appears

            if new:
                for word in self.words:
                    if word in new:
                        yield (word, new)

            else:
                time.sleep(0.5)


if __name__ == '__main__':
    print('LOG START (inner - as main).')
    main(sys.argv[1:])

编辑-1:更多想法

我还没有尝试过的三种变通方法是套接字(在这个答案中也建议了) 在python中使用多个控制台 ),通过进程ID获取进程对象以进行更多控制,并使用ctypes库直接访问Windows控制台API,允许设置屏幕缓冲区,因为控制台可以有多个缓冲区,但只有一个活动缓冲区(在 CreateConsoleScreenBuffer function ).

编辑-2:通过套接字解决方案(概念验证)

使用套接字作为解决方法,以便显示新的日志实体,正如在 ,在一般情况下也是有效的。尽管如此,对于某些东西来说,这似乎太费劲了,应该简单地发送到接收控制台的进程。

班级:

class sockettest(threading.Thread):

    def __init__(self, host, port, file):
        self.host = host
        self.port = port
        self.file = file
        threading.Thread.__init__(self)

    def run(self):
        print('run sockettest')
        socketing(self.host, self.port, self.file)
        print('sockettest done')

职能:

def socketing(host, port, file):
    pexe = 'C:\Python34\python.exe '
    script = 'C:\\Users\\Public\\Documents\\test\test-005.py'
    hostk = '--address'
    hostv = str(host)
    portk = '--port'
    portv = str(port)
    filek = '--file'
    filev = file

    file = open(file, 'a')
    file.close()
    time.sleep(1)

    print('HOST START (outer): ' + pexe + script + ' ' + hostk + ' ' + hostv + ' ' + portk + ' ' + portv + ' ' + filek + ' ' + filev)
    proc = Popen([pexe, script, hostk, hostv, portk, portv, filek, filev], universal_newlines=True, creationflags=CREATE_NEW_CONSOLE)

    print('HOST FINISH (outer): ' + pexe + script + ' ' + hostk + ' ' + hostv + ' ' + portk + ' ' + portv + ' ' + filek + ' ' + filev)

    time.sleep(2)

电话:

# The file tempdata is filled with several strings of "0\n1\n2\n"
# Looking like this:
# 0
# 1
# 2
# 0
# 1
# 2

proxyfile = 'C:\\Users\\Public\\Documents\\test\\tempdata'
f = open(proxyfile, 'a')
f.close()
time.sleep(1)

thread = sockettest('127.0.0.1', 8888, proxyfile)
thread.start()
thread.join(REPETITIONS * REPETITIONS)

以下脚本基于 Python: Socket programming server-client application using threads . 在这里,我只保留了 类命令行(Cmd)

import socket
import sys
import threading
import time
from cmd import Cmd
from argparse import ArgumentParser
from queue import Queue

BUFFER_SIZE = 5120

class CommandLine(Cmd):
    """Custom console"""

    def __init__(self, fn, words, queue):
        Cmd.__init__(self)
        self.fn = fn
        self.words = words
        self.queue = queue

    def watch(self):
        fp = open(self.fn, 'r')
        while True:
            time.sleep(0.05)
            new = fp.readline()

            # Once all lines are read this just returns ''
            # until the file changes and a new line appears
            self.queue.put_nowait(new)


def main(args):
    parser = ArgumentParser(description="Parse arguments.")
    parser.add_argument("-a", "--address", type=str, default='127.0.0.1', required=False)
    parser.add_argument("-p", "--port", type=str, default='8888', required=False)
    parser.add_argument("-f", "--file", type=str, default='', required=False)
    arguments = parser.parse_args(args)

    if not arguments.address:
        print('HOST PRE-START (inner): host argument not found. Creating new default entry.')
        arguments.host = '127.0.0.1'
    if not arguments.port:
        print('HOST PRE-START (inner): port argument not found. Creating new default entry.')
        arguments.port = '8888'
    if not arguments.file:
        print('HOST PRE-START (inner): file argument not found. Creating new default entry.')
        arguments.file = 'C:\\Users\\Public\\Documents\\test\\tempdata'

    file_queue = Queue()

    print('HOST START (inner): ' + ' ' + arguments.address + ':' + arguments.port + ' --file ' + arguments.file)

    # Start server
    thread = threading.Thread(target=start_server, args=(arguments.address, arguments.port, ))
    thread.start()
    time.sleep(1)

    # Start client
    thread = threading.Thread(target=start_client, args=(arguments.address, arguments.port, file_queue, ))
    thread.start()

    # Start file reader
    f = open(arguments.file, 'a')
    f.close()
    time.sleep(1)

    words = ['word']
    console = CommandLine(arguments.file, words, file_queue)
    console.prompt = ''

    thread = threading.Thread(target=console.cmdloop, args=('', ))
    thread.start()
    print("\n")

    for hit_word, hit_sentence in console.watch():
        print("Found %r in line: %r" % (hit_word, hit_sentence))

    print('HOST FINISH (inner): ' + ' ' + arguments.address + ':' + arguments.port)


def start_client(host, port, queue):
    host = host
    port = int(port)         # arbitrary non-privileged port
    queue = queue

    soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    try:
        soc.connect((host, port))
    except:
        print("Client connection error" + str(sys.exc_info()))
        sys.exit()

    print("Enter 'quit' to exit")
    message = ""

    while message != 'quit':
        time.sleep(0.05)
        if(message != ""):
            soc.sendall(message.encode("utf8"))
            if soc.recv(BUFFER_SIZE).decode("utf8") == "-":
                pass        # null operation

        string = ""
        if (not queue.empty()):
            string = str(queue.get_nowait()) + "\n"

        if(string == None or string == ""):
            message = ""
        else:
            message = string

    soc.send(b'--quit--')


def start_server(host, port):
    host = host
    port = int(port)         # arbitrary non-privileged port

    soc = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # SO_REUSEADDR flag tells the kernel to reuse a local socket in TIME_WAIT state, without waiting for its natural timeout to expire
    soc.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    print("Socket created")

    try:
        soc.bind((host, port))
    except:
        print("Bind failed. Error : " + str(sys.exc_info()))
        sys.exit()

    soc.listen(5)       # queue up to 5 requests
    print("Socket now listening")

    # infinite loop- do not reset for every requests
    while True:
        connection, address = soc.accept()
        ip, port = str(address[0]), str(address[1])
        print("Connected with " + ip + ":" + port)

        try:
            threading.Thread(target=client_thread, args=(connection, ip, port)).start()
        except:
            print("Thread did not start.")
            traceback.print_exc()

    soc.close()


def client_thread(connection, ip, port, max_buffer_size=BUFFER_SIZE):
    is_active = True

    while is_active:
        client_input = receive_input(connection, max_buffer_size)

        if "--QUIT--" in client_input:
            print("Client is requesting to quit")
            connection.close()
            print("Connection " + ip + ":" + port + " closed")
            is_active = False
        elif not client_input == "":
            print("{}".format(client_input))
            connection.sendall("-".encode("utf8"))
        else:
            connection.sendall("-".encode("utf8"))


def receive_input(connection, max_buffer_size):
    client_input = connection.recv(max_buffer_size)
    client_input_size = sys.getsizeof(client_input)

    if client_input_size > max_buffer_size:
        print("The input size is greater than expected {}".format(client_input_size))

    decoded_input = client_input.decode("utf8").rstrip()  # decode and strip end of line
    result = process_input(decoded_input)

    return result


def process_input(input_str):
    return str(input_str).upper()


if __name__ == '__main__':
    print('HOST START (inner - as main).')
    main(sys.argv[1:])

直接控制子流程的控制台输入管道/缓冲区将是解决此问题的首选方案。因为这是500人的荣誉。

不幸的是我没时间了。因此,我现在可能会使用其中一种变通方法,并在以后用适当的解决方案替换它们。或者我必须使用nuclear选项,只有一个控制台,在任何用户键盘输入期间暂停正在进行的日志输出,然后打印。当然,这可能会导致缓冲区问题,当用户决定在中途键入某些内容时。

编辑-3:包含接受答案的代码(一个文件)

根据James Kent的回答,当我通过Windows命令行(cmd)或PowerShell用代码启动脚本时,我得到了所需的行为。但是,当我通过Eclipse/PyDev和“Python run”启动同一个脚本时,输出总是打印在主Eclipse/PyDev控制台上,而子流程的第二个控制台保持为空并处于非活动状态。不过,我想这是另一个系统/环境特性,是另一个问题。

from sys import argv, stdin, stdout
from threading import Thread
from cmd import Cmd
from time import sleep
from datetime import datetime
from subprocess import Popen, PIPE, CREATE_NEW_CONSOLE

INTRO = 'command line'
PROMPT = '> '


class CommandLine(Cmd):
    """Custom console"""

    def __init__(self, subprocess, intro=INTRO, prompt=PROMPT):
        Cmd.__init__(self)
        self.subprocess = subprocess
        self.intro = intro
        self.prompt = prompt
        self.doc_header = intro
        self.running = False

    def do_date(self, args):
        """Prints the current date and time."""
        print(datetime.now())
        sleep(1)

    def do_exit(self, args):
        """Exits this command line application."""
        print("Exit by user command.")
        if self.subprocess is not None:
            try:
                self.subprocess.terminate()
            except:
                self.subprocess.kill()
        exit()


class Console():

    def __init__(self):
        if '-r' not in argv:
            self.p = Popen(
                ['python.exe', __file__, '-r'],
                stdin=PIPE,
                creationflags=CREATE_NEW_CONSOLE
            )
        else:
            while True:
                data = stdin.read(1)
                if not data:
                    #                     break
                    sleep(1)
                    continue
                stdout.write(data)

    def write(self, data):
        self.p.stdin.write(data.encode('utf8'))
        self.p.stdin.flush()

    def getSubprocess(self):
        if self.p:
            return self.p
        else:
            return None


class Feeder (Thread):

    def __init__(self, console):
        self.console = console
        Thread.__init__(self)

    def run(self):
        feeding(self.console)


def feeding(console):
    for i in range(0, 100):
        console.write('test %i\n' % i)
        sleep(1)


if __name__ == '__main__':
    p = Console()
    if '-r' not in argv:
        thread = Feeder(p)
        thread.setDaemon(True)
        thread.start()

        cl = CommandLine(subprocess=p.getSubprocess())
        cl.use_rawinput = False
        cl.prompt = PROMPT
        cl.cmdloop('\nCommand line is waiting for user input (e.g. help).')

编辑-3:荣誉提名

在上面的问题文本中,我提到了使用ctypes库直接访问Windows控制台API作为另一种解决方法(在“编辑-1:更多想法”下)。或者在某种程度上只使用一个控制台,输入提示符始终位于底部,作为解决整个问题的核心选项(在“编辑-2:进一步思考”下)

对于使用ctypes库,我会将自己定位在以下问题的答案上 Change console font in Windows Keep console input line below output . 我认为这两个答案都可能提供关于这个问题的潜在merrit,也许它们对其他人如何通过这篇文章有帮助。此外,如果我有时间,我会尝试一下,如果他们能以某种方式工作。

1 回复 | 直到 7 年前

SandPiper 7 年前

您遇到的问题是Windows上控制台子系统的体系结构,您通常看到的控制台窗口不是由cmd.exe托管的,而是由conhost.exe托管的。conhost窗口的子进程只能连接到单个conhost实例,这意味着每个进程只能有一个窗口。

然后,这会导致为您想要的每个控制台窗口都有一个额外的进程,然后为了查看在该窗口中显示的任何内容,您需要查看stdin和stdout通常是如何处理的,因为它们是由conhost实例写入和读取的,除非您将stdin转换为管道(这样您就可以写入进程),它不再来自conhost,而是来自您的父进程,因此conhost对它没有可见性。这意味着写入stdin的任何内容仅由子进程读取,因此conhost不会显示。

如果将stdin设置为管道,则所有发送到新控制台窗口的键盘输入都将无效,因为stdin未连接到该窗口。

这里有一个尝试:

#!python3

import sys, subprocess, time

class Console():
    def __init__(self):
        if '-r' not in sys.argv:
            self.p = subprocess.Popen(
                ['python.exe', __file__, '-r'],
                stdin=subprocess.PIPE,
                creationflags=subprocess.CREATE_NEW_CONSOLE
                )
        else:
            while True:
                data = sys.stdin.read(1)
                if not data:
                    break
                sys.stdout.write(data)

    def write(self, data):
        self.p.stdin.write(data.encode('utf8'))
        self.p.stdin.flush()

if (__name__ == '__main__'):
    p = Console()
    if '-r' not in sys.argv:
        for i in range(0, 100):
            p.write('test %i\n' % i)
            time.sleep(1)

因此,在两个进程之间建立一个简单的管道,并将输入返回到输出(如果是子进程),我使用了-r来表示实例是否是一个进程,但还有其他方法取决于您如何实现它。

写入stdin后需要刷新,因为python通常使用缓冲。
这种方法的编写方式旨在将其放在自己的模块中,因此使用 __file__
由于使用了 __文件__

编辑1

对于可以使用cx\U冻结的版本:

Console.py

import sys, subprocess

class Console():
    def __init__(self, ischild=True):
        if not ischild:
            if hasattr(sys, 'frozen'):
                args = ['Console.exe']
            else:
                args = [sys.executable, __file__]
            self.p = subprocess.Popen(
                args,
                stdin=subprocess.PIPE,
                creationflags=subprocess.CREATE_NEW_CONSOLE
                )
        else:
            while True:
                data = sys.stdin.read(1)
                if not data:
                    break
                sys.stdout.write(data)

    def write(self, data):
        self.p.stdin.write(data.encode('utf8'))
        self.p.stdin.flush()

if (__name__ == '__main__'):
    p = Console()

test.py

from Console import Console
import sys, time

if (__name__ == '__main__'):
    p = Console(False)
    for i in range(0, 100):
        p.write('test %i\n' % i)
        time.sleep(1)

setup.py

from cx_Freeze import setup, Executable

setup(
    name = 'Console-test',
    executables = [
        Executable(
            'Console.py',
            base=None,
            ),
        Executable(
            'test.py',
            base=None,
            )
        ]
)

编辑2

新版本应该在开发工具(如IDLE)下工作

#!python3

import ctypes, sys, subprocess

Kernel32 = ctypes.windll.Kernel32

class Console():
    def __init__(self, ischild=True):
        if ischild:
            # try allocate new console
            result = Kernel32.AllocConsole()
            if result > 0:
                # if we succeed open handle to the console output
                sys.stdout = open('CONOUT$', mode='w')
        else:
            # if frozen we assume its names Console.exe
            # note that when frozen 'Win32GUI' must be used as a base
            if hasattr(sys, 'frozen'):
                args = ['Console.exe']
            else:
                # otherwise we use the console free version of python
                args = ['pythonw.exe', __file__]
            self.p = subprocess.Popen(
                args,
                stdin=subprocess.PIPE
                )
            return
        while True:
            data = sys.stdin.read(1)
            if not data:
                break
            sys.stdout.write(data)

    def write(self, data):
        self.p.stdin.write(data.encode('utf8'))
        self.p.stdin.flush()

if (__name__ == '__main__'):
    p = Console()

test.py

从控制台导入控制台

如果(uuuu name_uuuuuu=='uuuuuu main_uuuuu'):
p=控制台(错误)
时间。睡眠(1)

setup.py

from cx_Freeze import setup, Executable

setup(
    name = 'Console-test',
    executables = [
        Executable(
            'Console.py',
            base='Win32GUI',
            ),
        Executable(
            'test.py',
            base=None,
            )
        ]
)

这可以变得更加健壮,即始终检查现有控制台,如果在创建新控制台之前发现,则将其分离,并且可能更好地处理错误。

Goldfish Doc 5 年前

因为您在windows上,所以可以使用Win32 console模块为线程或子流程输出打开第二个或多个控制台。如果您在windows上,这是最简单、最容易的方法。

下面是一个示例代码:

import win32console
import multiprocessing

def subprocess(queue):
    win32console.FreeConsole() #Frees subprocess from using main console
    win32console.AllocConsole() #Creates new console and all input and output of subprocess goes to this new console
    while True:
        print(queue.get())
        #prints any output produced by main script passed to subprocess using queue

if __name__ == "__main__": 
    queue = multiprocessing.Queue()
    multiprocessing.Process(target=subprocess, args=[queue]).start()
    while True:
        print("Hello World in main console")
        queue.put("Hello work in sub process console")
        #sends above string to subprocess and it prints it into its console

        #and whatever else you want to do in ur main process

这是你的电话号码 win32console module documentation

具有实时输入和多个控制台的Python子流程

略为不同的方法

更多信息

更多代码

编辑-1:通过文件解决方案(概念验证)

编辑-1:更多想法

编辑-2:通过套接字解决方案(概念验证)

编辑-3:包含接受答案的代码(一个文件)

编辑-3:荣誉提名

编辑1

Console.py

test.py

setup.py

编辑2

test.py

setup.py