python에서 현재 콘솔에있는 행을 읽는 방법?

Question

asyncio 모듈과 함께 콘솔에 입력 한 내용을 읽을 수있는 코드 줄을 사용하고 싶습니다. 내 코드는 서버에서 데이터를받을 때 데이터를 출력하고, 그 후에는 입력 한 내용을 읽고 변수에 저장하려고합니다. 비표준 모듈을 사용해야하는 것은 괜찮습니다. 프로그램에 대한

정보 :

• 는 현재 다음, 표준 입력이
• 프로그램 loop.create_connection를 사용하여 실행 asyncio loop.run_in_executor를 사용하여 비 차단 읽는 작업으로 추가 된 표준 입력 리더 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

나는 하나의 문자와 몇 가지 코드 변경을 잡으려고 msvcrt을 사용하여 문제를 (Windows 만 해당) 해결했다.

새로운 기능/변경

1. draw_screen :

   • 인쇄 (랙이 얻을 수) 나는 정보를 인쇄 할 때 내가 저장 os.system("cls")
   • 인쇄 라인을 사용
   • 명확한 화면 현재 입력하고있는 커서 (">> Hello")

2. 쓰기 : msvcrt.getch와

   • 교체 sys.stdin.readline
   • 더 많은 변화
   사방

3. :

   • 때마다 나는 안으로 인쇄 할 것입니다. 대신 모든 것을 보유하는 배열 인 screen 변수에이를 추가하십시오. 그런 다음 호출 draw_screen()
   • 는
   • 이 다른 print 문
   (지금은)

최종 코드 제거 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()