xticks barplot python input console subprocess real-time

barplot - Subproceso Python con entrada en tiempo real y múltiples consolas



pandas plot (1)

El problema al que te enfrentas es la arquitectura del subsistema de consola en Windows, la ventana de consola que normalmente ves no está alojada en cmd.exe, sino que conhost.exe, un proceso secundario de una ventana de conhost solo puede conectarse a una Conhost instancia única significa que está limitado a una sola ventana por proceso.

Esto lleva a tener un proceso adicional para cada ventana de consola que desea tener, y para ver cómo se muestra en esa ventana, debe ver cómo se manejan normalmente la entrada estándar y la salida estándar, ya que se escriben y leen En la instancia de Conhost, excepto que si convierte la entrada estándar en una canalización (para poder escribir en el proceso), ya no proviene de Conhost sino de su proceso principal y, por lo tanto, Conhost no tiene visibilidad. Esto significa que cualquier cosa escrita en stdin solo es leída por el proceso hijo, por lo que no es mostrada por conhost.

Que yo sepa, no hay una manera de compartir la tubería de esa manera.

Como efecto secundario, si convierte la entrada estándar en una canalización, todas las entradas del teclado enviadas a la nueva ventana de la consola no van a ninguna parte, ya que la entrada estándar no está conectada a esa ventana.

Para una función de salida solamente, esto significa que puede generar un nuevo proceso que se comunique con el padre a través de una canalización a la entrada estándar y que todo se transmita a la salida estándar.

Heres un intento:

#!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)

Así que una canalización simple y agradable entre dos procesos y haciendo eco de la entrada a la salida si es el subproceso, utilicé a -r para indicar si la instancia es un proceso, pero hay otras formas en función de cómo se implementa.

Varias cosas a tener en cuenta:

  • la descarga después de escribir en stdin es necesaria, ya que Python normalmente usa buffering.
  • La forma en que se escribe este enfoque está orientada a estar en su propio módulo, por lo tanto, se usa __file__
  • debido al uso de __file__ este enfoque puede necesitar modificaciones si se congela usando cx_Freeze o similar.

EDITAR 1

para una versión que se puede congelar con cx_Freeze:

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, ) ] )

Editar 2

Nueva versión que debería funcionar bajo herramientas dev como IDLE

Console.py

#!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

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=''Win32GUI'', ), Executable( ''test.py'', base=None, ) ] )

Esto podría hacerse más robusto, es decir, siempre buscando una consola existente y separándola si se encuentra antes de crear una nueva consola, y posiblemente un mejor manejo de errores.

El problema principal

En pocas palabras: quiero dos consolas para mi programa. Uno para la entrada de usuario activo. Y la otra para salida de log pura. (El código de trabajo que incluye la respuesta aceptada se encuentra en el texto de la pregunta a continuación, en la sección "Edición-3". Y en la sección "Edición-1" y la sección "Edición-2" funcionan como soluciones provisionales).

Para esto tengo un script de Python de línea de comando principal, que se supone que abre una consola adicional solo para la salida del registro. Para esto, tengo la intención de redirigir la salida del registro, que se imprimiría en la consola del script principal, al stdin de la segunda consola, que comienzo como un subproceso. (Uso subproceso, porque no encontré ninguna otra forma de abrir una segunda consola).

El problema es que parece que puedo enviar al stdin de esta segunda consola, sin embargo, nada se imprime en esta segunda consola.

A continuación se muestra el código que utilicé para experimentar (con Python 3.4 en PyDev en Windows 10). La función de writing(input, pipe, process) contiene la parte, donde la cadena generada se copia a la pipe estándar pasada, de la consola abierta a través de subproceso. La función escritura (...) se ejecuta a través de la clase writetest(Thread) . (Dejé un código, que comenté).

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")

Un enfoque ligeramente diferente

El siguiente fragmento de código adicional funciona en lo que respecta a extraer la salida estándar del subproceso. Sin embargo, el stdin enviado anteriormente todavía no se imprime en la segunda consola. Además, la segunda consola se cierra de inmediato.

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")

Más información

Tan pronto como uso uno de los parámetros stdin=data_pipe, stdout=PIPE, stderr=PIPE para iniciar el subproceso, la segunda consola resultante no está activa y no acepta entrada de teclado (lo que no es deseable, aunque podría ser información útil aquí).

El método de subproceso communicate() no se puede usar para esto ya que espera a que finalice el proceso.

Más código

Finalmente el código para el archivo, que es para la segunda consola.

C: / Users / Public / Documents / test / test-cmd.py

from cmd import Cmd from time import sleep from datetime import datetime INTRO = ''command line'' PROMPT = ''> '' class CommandLine(Cmd): """Custom console""" def __init__(self, intro=INTRO, prompt=PROMPT): Cmd.__init__(self) self.intro = intro self.prompt = prompt self.doc_header = intro self.running = False def do_dummy(self, args): """Runs a dummy method.""" print("Do the dummy.") self.running = True while(self.running == True): print(datetime.now()) sleep(5) def do_stop(self, args): """Stops the dummy method.""" print("Stop the dummy, if you can.") self.running = False def do_exit(self, args): """Exits this console.""" print("Do console exit.") exit() if __name__ == ''__main__'': cl = CommandLine() cl.prompt = PROMPT cl.cmdloop(INTRO)

Pensamientos

Hasta ahora ni siquiera estoy seguro de si la interfaz de línea de comandos de Windows ofrece la capacidad de aceptar otra entrada que no sea la del teclado (en lugar de la tubería estándar deseada o similar). Aunque, teniendo algún tipo de modo pasivo, lo espero.

¿Por qué esto no funciona?

Edit-1: Solución a través del archivo (prueba de concepto)

Usar un archivo como solución temporal para mostrar su contenido nuevo, como se sugiere en la respuesta de Trabajar varias consolas en Python , funciona en general. Sin embargo, dado que el archivo de registro crecerá a muchos GB, no es una solución práctica en este caso. Al menos requeriría la división de archivos y su manejo adecuado.

La clase:

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'')

La función:

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)

La llamada:

# 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)

El script de cola ("test-004.py"):

Como Windows no ofrece el comando tail, utilicé el siguiente script ( basado en la respuesta de ¿Cómo implementar un equivalente pythonic de tail -F? ), Que funcionó para esto. La clase de línea de comandos adicional, aunque innecesaria class CommandLine(Cmd) fue inicialmente un intento de mantener abierta la segunda consola (porque faltaba el argumento del archivo de script). Sin embargo, también demostró ser útil para mantener la consola imprimiendo con fluidez el nuevo contenido del archivo de registro. De lo contrario, la salida no fue determinista / predecible.

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:])

Edit-1: Más pensamientos

Tres soluciones alternativas, que no probé todavía y que podrían funcionar son los sockets (también sugeridos en esta respuesta Trabajando múltiples consolas en python ), obteniendo un objeto de proceso a través del ID de proceso para un mayor control, y usando la biblioteca ctypes para acceder directamente a Windows API de la consola, que permite configurar el búfer de pantalla, ya que la consola puede tener varios búferes, pero solo un búfer activo (como se indica en los comentarios de la documentación para la función CreateConsoleScreenBuffer ).

Sin embargo, el uso de sockets puede ser el más fácil. Y al menos el tamaño del registro no importa de esta manera. Sin embargo, los problemas de conexión pueden ser un problema aquí.

Edit-2: Solución a través de sockets (prueba de concepto)

El uso de sockets como solución temporal para mostrar nuevas entidades de registro, como también se sugirió en la respuesta de Trabajar múltiples consolas en python , también funciona en general. Sin embargo, esto parece ser demasiado esfuerzo para algo, que simplemente debe enviarse al proceso de la consola receptora.

La clase:

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'')

La función:

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)

La llamada:

# 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)

El script de socket ("test-005.py"):

La siguiente secuencia de comandos se basa en Python: aplicación de cliente-servidor de programación de sockets que utiliza subprocesos . Aquí acabo de mantener la class CommandLine(Cmd) como generador de entrada de registro. En este punto, no debería ser un problema, poner al cliente en el script principal, que llama a la segunda consola y luego alimenta la cola con registros reales en lugar de (nuevas) líneas de archivos. (El servidor es la impresora.)

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:])

Edit-2: Además pensamientos

Tener el control directo de la entrada / consola de la consola del subproceso sería la solución preferible a este problema. Porque esta es la recompensa de 500 Reputación.

Lamentablemente me estoy quedando sin tiempo. Por lo tanto, podría usar una de esas soluciones por ahora y reemplazarla con la solución adecuada más adelante. O tal vez tenga que usar la opción nuclear, solo una consola, donde la salida del registro en curso se detiene durante cualquier entrada del teclado del usuario y luego se imprime. Por supuesto, esto podría llevar a problemas en el búfer, cuando el usuario decide escribir algo a la mitad.

Edit-3: Código que incluye la respuesta aceptada (un archivo)

Con la respuesta de James Kent obtengo el comportamiento deseado, cuando comienzo un script con el código a través de la línea de comandos de Windows (cmd) o PowerShell. Sin embargo, cuando comienzo este mismo script a través de Eclipse / PyDev con "Python run", la salida siempre se imprime en la consola principal de Eclipse / PyDev, mientras que la segunda consola del subproceso permanece vacía y permanece inactiva. Sin embargo, creo que esta es otra especialidad de sistema / entorno y un tema diferente.

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).'')

Edit-3: Menciones honoríficas

En el texto de las preguntas anteriores, mencioné el uso de la biblioteca ctypes para acceder directamente a la API de la consola de Windows como otro entorno de trabajo (en "Editar-1: Más ideas"). O usando solo una consola de forma que el mensaje de entrada siempre se mantenga en la parte inferior como opción nuclear para todo este problema. (en "Editar-2: Además pensamientos")

Para usar la biblioteca ctypes me habría orientado en la siguiente respuesta a Cambiar la fuente de la consola en Windows . Y para usar una sola consola, habría intentado la siguiente respuesta a Mantener la línea de entrada de la consola debajo de la salida . Creo que estas dos respuestas pueden ofrecer un posible error con respecto a este problema y tal vez sean útiles para los demás. Además, si encuentro el tiempo, lo intentaré si funcionan de alguna manera.