Как читать строку из python в настоящее время в консоли?

Question

Я хочу использовать строку кода, которая может читать то, что я набираю в консоли, для использования с модулем asyncio в python. Мой код печатает данные, когда он получает его с сервера, и после этого я хочу, чтобы он прочитал то, что я печатал, и сохранил его в переменной. Я в порядке с использованием нестандартного модуля.

Информация о программе:

• В настоящее время читает STDIN неблокирующим с использованием asyncio loop.run_in_executor
• Программа работает с использованием loop.create_connection, а затем использует loop.run_forever с читателем стандартного ввода добавленным в качестве задачи.

Код:

#!python3.5 
#chatroom client 

import socket, sys, os, traceback, asyncio 
from threading import Lock 

DBG = False #More Error printing using full_error function 

#async stuf 
loop = asyncio.get_event_loop() 
lock = Lock() 
prev_stdin = "" 

#Server Location 
target = 'localhost' 
port = 17532 
buffer_size = 1024 
server = None 
transports = [] #an array of async.iotransport`s 

#Functions for Error handling 
full_error = lambda: traceback.print_exception(*sys.exc_info()) 
pause = lambda: None #make it global for below 

#Pause system compatability 
if os.name[0:5]=='posix': 
    def pause(): 
     os.system('read -n1 -r -p "Press any key to continue . . ." key') 
elif os.name[0:2]=='nt': 
    def pause(): 
     os.system("pause") 
else: 
    def pause(): 
     input("Press enter to continue . . .") 

def empty_stdin(): 
    output = "" 
    while True: 
     temp = sys.stdin.buffer.read(1) 
     if temp == "": 
      break 
     else: 
      output += temp 
    return output 

if DBG: 
    loop.set_debug(DBG) 

async def write(): 
    loop = asyncio.get_event_loop() 
    while True: 
     line = await loop.run_in_executor(None, sys.stdin.readline) 
     line = prev_stdin + line 
     prev_stdin = "" 
     for i in transports: 
      loop.call_soon(i.write,line.encode('utf-8')) 

class Server(asyncio.Protocol): 
    def connection_made(self,transport): 
     self.transport = transport 
     transport.write(b"Connected") 
    def data_received(self, data): 
     print("\n%s"%data.decode('utf-8')) 
     prev_stdin = empty_stdin() 
     print(prev_stdin,) 



try: 
    coro = loop.create_connection(Server, host=target, port=port) 
    task = loop.create_task(coro) 
    print("hi") 
    print(transports) 
    server, serverp = loop.run_until_complete(task) 
    transports += [server] 
    user_input = loop.create_task(write()) 
    #loop.add_reader(sys.stdin, write) 
    loop.run_forever() 
    print('end') 

except Exception as e: 
    if DBG: 
     full_error() 
    else: 
     print(type(e)) 
     print(e) 
    pause() 
    sys.exit() 

Answer 1

Я решил проблему (только для Windows) с помощью msvcrt поймать одного символа и некоторые изменения кода.

Новые функции/Изменения

1. draw_screen:

   • очистить экран с помощью os.system("cls")
   • печати линий, которые спасают меня, когда я печатаю информацию (удалось получить лага)
   • печати курсор с тем, что я набираю в настоящее время («>> Hello»)

2. пишут:

   • заменен sys.stdin.readline с msvcrt.getch
   • Много больше изменений

3. везде:

   • Всякий раз Я бы напечатал в добавьте его в переменную screen, которая представляет собой массив, который содержит все. Затем она вызывает draw_screen()
   • удалены prev_stdin материал
   • удалены некоторые другие операторы печати

Окончательный код (сейчас):

#!python3.5 
#chatroom client 

import sys, os, traceback, asyncio, msvcrt 
from threading import Lock 

DBG = False #More Error printing using full_error function 

screen = [] #holds lines of text that are printed to the screen 
line = "" 

#async stuf 
loop = asyncio.get_event_loop() 
lock = Lock() 


#Server Location 
target = 'localhost' 
port = 17532 
buffer_size = 1024 
server = None 
transports = [] #an array of async.iotransport`s 

#Functions for Error handling 
full_error = lambda: traceback.print_exception(*sys.exc_info()) 

#system compatable functions 
if os.name[0:5]=='posix': 
    def pause(): 
     os.system('read -n1 -r -p "Press any key to continue . . ." key') 
    def cls(): 
     os.system("clear") 
elif os.name[0:2]=='nt': 
    def pause(): 
     os.system("pause") 
    def cls(): 
     os.system("cls") 
else: 
    def pause(): 
     input("Press enter to continue . . .") 
    def cls(): 
     os.system("clear") 

def draw_screen(): 
    cls() 
    for i in screen: 
     print(i) 
    print("> ", line, end="\r") #Draws current line of text being typed 

if DBG: 
    loop.set_debug(DBG) 

async def write(): 
    global line 
    while True: 
     temp = await loop.run_in_executor(None, msvcrt.getch) 
     temp = temp.decode('utf-8') 
     if temp == "\b": 
      line = line[0:len(line)-1] 
     elif not (temp == "\r" or temp == "\n"): 
      line += temp 
     else: 
      for i in transports: 
       loop.call_soon(i.write,line.encode('utf-8')) 
      line = "" 
     draw_screen() 

class Server(asyncio.Protocol): 
    def connection_made(self,transport): 
     self.transport = transport 
     transport.write(b"Connected") 
    def data_received(self, data): 
     global screen 
     screen += ["%s"%data.decode('utf-8')] 
     draw_screen() 



try: 
    coro = loop.create_connection(Server, host=target, port=port) 
    task = loop.create_task(coro) 
    server, serverp = loop.run_until_complete(task) 
    transports += [server] 
    user_input = loop.create_task(write()) 
    loop.run_forever() 
    print('end') 

except Exception as e: 
    if DBG: 
     full_error() 
    else: 
     print(type(e)) 
     print(e) 
    pause() 
    sys.exit()