如何在python中创建后台无阻塞输入循环? [英] How can I make a background, non-blocking input loop in python?

问题描述

因此，我有一个正在使用的小型控制台应用程序，该应用程序运行一个网络抓取过程，并且我希望能够在执行过程中向其提供控制台命令以对其进行控制.为此，我将需要某种形式的非阻塞键盘输入，因为该程序可能由于意外错误而自行终止，并且我不希望某些线程挂起并在终止时等待输入.

So, I have a small console app I'm working on that runs a web-scraping process, and I want to be able to give it console commands mid-execution to control it. To do this I'll need some form of non-blocking keyboard input, as the program may self-terminate due to unexpected errors, and I don't want some thread hanging about and waiting for input when termination has occurred.

我已经散列了以下内容:

I have the following already hashed out:

import threading
import time
import queue

input_queue = queue.Queue()
command_input_event = threading.Event()

def kbdListener():
    global input_queue, command_input_event
    kbdInput = ''
    while kbdInput.lower() not in ['quit', 'exit', 'stop']:
        kbdInput = input("> ")
        input_queue.put(kbdInput)
        command_input_event.set()
        input_queue.join()

listener = threading.Thread(target=kbdListener)
listener.start()
stop = False
while not stop:
    if command_input_event.is_set():
        while not input_queue.empty():
            command = input_queue.get()
            if command.lower() in ['quit', 'exit', 'stop']:
                print('Stopping')
                while not input_queue.empty():
                    input_queue.get()
                    input_queue.task_done()
                input_queue.task_done()
                stop = True
                break
            else:
                print('Command "{}" received and processed'.format(command))
                input_queue.task_done()

我的问题是，在行while not stop:上，将在我的程序中检查另一种情况，该条件确定主循环是否已终止.如果发生这种情况，则主线程将停止，但后台listener线程仍将在等待输入；否则，将继续运行.我想避免的情况.

My problem is that on the line while not stop: there will be another condition being checked in my program, that determines if the main loop has terminated. If this eventuality was to occur then the main thread would stop, but the background listener thread would still be waiting for input; the situation I'm trying to avoid.

我不受这种方法的束缚，因此，如果有一些获取非阻塞输入的替代方法，那么我也乐于接受该建议.

I'm not tied in to this approach, so if there is some alternative method for getting a non-blocking input, then I would be open to that advice as well.

推荐答案

因此，在对此进行一些工作之后，我想到了以下内容，该内容允许后台线程输入，然后由前台线程处理，但完全是非阻塞并根据用户实时输入进行更新.

So after some work on this I came up with the following, which allows for a background thread input, which is then processed by the foreground thread, but is entirely non-blocking and updates as the user types in real-time.

import threading
import queue
import sys
from msvcrt import getch, kbhit

"""
Optional extra that I found necessary to get ANSI commands working on windows:

import colorama
colorama.init()
"""

class ClassWithNonBlockingInput:

    def __init__(self):
        self.command_queue = queue.Queue()
        self.command_available_event = threading.Event()
        self.stop_event = threading.Event()

    def run(self):
        main_loop = threading.Thread(target=self.main_loop)
        main_loop.start()
        listener = threading.Thread(target=self.keyboard_listener)
        listener.start()
        stop = False
        while main_loop.is_alive() and not stop:
            if self.command_available_event.is_set():
                while not self.command_queue.empty():
                    command = self.command_queue.get()
                    #Process command here, long jobs should be on a seperate thread.
                    self.command_queue.task_done()
                self.command_available_event.clear()

    def main_loop(self):
        #Main processing loop, may set self.stop_event at some point to terminate early.
        pass

    def keyboard_listener(self):
        line = []
        line_changed = False
        while not self.stop_event.is_set():
            while kbhit():
                c = getch()
                if c == b'\x00' or c == b'\xe0':
                    #This is a special function key such as F1 or the up arrow.
                    #There is a second identifier character to clear/identify the key.
                    id = getch()
                    #Process the control key by sending commands as necessary.
                elif c == b'\x08':
                    #Backspace character, remove last character.
                    if len(line) > 0:
                        line = line[:-1]
                    line_changed = True
                elif c == b'\x7f':
                    #ctrl-backspace, remove characters until last space.
                    while len(line) > 0 and line[-1] != b' ':
                        line = line[:-1]
                    line_changed = True
                elif c == b'\x1b':
                    #Escacpe key, process as necessary.
                    pass
                elif c == b'\r':
                    #Enter key, send command to main thread.
                    print()
                    command = b''.join(line).decode('utf-8')
                    self.command_queue.put(command)
                    self.command_available_event.set()
                    self.command_queue.join()
                    line = []
                    line_changed = True
                else:
                    #Append all other characters to the current line.
                    #There may be other special keys which need to be considered,
                    #  this is left as an exercise for the reader :P
                    line.append(c)
                    line_changed = True

                if line_changed:
                    #Clear the current output line
                    print('\033[2K', end='\r')
                    print(b''.join(line).decode('utf-8'), end='')
                    sys.stdout.flush()
                    line_changed = False

这应该为将来遇到此问题的任何人提供一个良好的开端.

This should give anyone who encounters this problem in the future, and stumbles upon this question a good head start.

这篇关于如何在python中创建后台无阻塞输入循环?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！