asyncio keyboard input that can be canceled

  • Last Update :
  • Techknowledgy :

Canceling the future has no effect because concurrent.futures.Future.cancel is a no-op once its callback has started executing. The best way to avoid the unwanted waiting is to avoid run_in_executor in the first place and just spawn your own thread:

async def ainput():
   loop = asyncio.get_event_loop()
fut = loop.create_future()
def _run():
   line = sys.stdin.readline()
loop.call_soon_threadsafe(fut.set_result, line)
threading.Thread(target = _run, daemon = True).start()
return await fut

The thread is created manually and marked as "daemon", so no one will wait for it at program shutdown. As a result, a variant of the code that uses ainput instead of run_in_executor(sys.stdin.readline) terminates as expected:

async def console_input_loop():
   while True:
   inp = await ainput()
print(f "[{inp.strip()}]")

# rest of the program unchanged

Suggestion : 2

An instance of asyncio.Handle is returned, which can be used later to cancel the callback.,An instance of asyncio.TimerHandle is returned which can be used to cancel the callback.,The event loop is the core of every asyncio application. Event loops run asynchronous tasks and callbacks, perform network IO operations, and run subprocesses.,This method can be used by servers that accept connections outside of asyncio but that use asyncio to handle them.

# will schedule "print("
Hello ", flush=True)"
   functools.partial(print, "Hello", flush = True))
import asyncio
import concurrent.futures

def blocking_io():
   # File operations(such as logging) can block the
# event loop: run them in a thread pool.
with open('/dev/urandom', 'rb') as f:

def cpu_bound():
   # CPU - bound operations will block the event loop:
   # in general it is preferable to run them in a
# process pool.
return sum(i * i
   for i in range(10 ** 7))

async def main():
   loop = asyncio.get_running_loop()

# # Options:

   # 1. Run in the
default loop 's executor:
result = await loop.run_in_executor(
   None, blocking_io)
print('default thread pool', result)

# 2. Run in a custom thread pool:
   with concurrent.futures.ThreadPoolExecutor() as pool:
   result = await loop.run_in_executor(
      pool, blocking_io)
print('custom thread pool', result)

# 3. Run in a custom process pool:
   with concurrent.futures.ProcessPoolExecutor() as pool:
   result = await loop.run_in_executor(
      pool, cpu_bound)
print('custom process pool', result)
srv = await loop.create_server(...)

async with srv:
   # some code

# At this point, srv is closed and no longer accepts new connections.
async def client_connected(reader, writer):
   # Communicate with the client with
# reader / writer streams.For example:
   await reader.readline()

async def main(host, port):
   srv = await asyncio.start_server(
      client_connected, host, port)
await srv.serve_forever()'', 0))
import asyncio
import selectors

selector = selectors.SelectSelector()
loop = asyncio.SelectorEventLoop(selector)

Suggestion : 3

We can do this using the 'input hook' feature of the interactive prompt. This hook is called by the prompt as it waits for the user to type (even for a fast typist the prompt is mostly waiting for the human to think and move their fingers). Although the details vary between prompts the logic is roughly,Matplotlib supports rich interactive figures by embedding figures into a GUI window. The basic interactions of panning and zooming in an Axes to inspect your data is 'baked in' to Matplotlib. This is supported by a full mouse and keyboard event handling system that you can use to build sophisticated interactive graphs.,These examples are aggressively dropping many of the complexities that must be dealt with in the real world such as keyboard interrupts, timeouts, bad input, resource allocation and cleanup, etc.,This gives us the illusion of simultaneously having interactive GUI windows and an interactive prompt. Most of the time the GUI event loop is running, but as soon as the user starts typing the prompt takes over again.

exec_count = 0
while True:
   inp = input(f "[{exec_count}] > ") # Read
ret = eval(inp) # Evaluate
print(ret) # Print
exec_count += 1 # Loop = True)
fig, ax = plt.subplots()
ln, = ax.plot([], [])

while True:
   x, y = get_new_data()
ln.set_data(x, y)
import time
import matplotlib.pyplot as plt
import numpy as np

fig, ax = plt.subplots()
th = np.linspace(0, 2 * np.pi, 512)
ax.set_ylim(-1.5, 1.5)

ln, = ax.plot(th, np.sin(th))

def slow_loop(N, ln):
   for j in range(N):
   time.sleep(.1) # to simulate some work

slow_loop(100, ln)
def slow_loop(N, ln):
   for j in range(N):
   time.sleep(.1) # to simulate some work
if j % 10:
   ln.set_ydata(np.sin(((j // 10) % 5 * th)))


               slow_loop(100, ln)
def callback(self: Artist, val: bool) - > None:

Suggestion : 4

Last Updated : 02 Dec, 2020

For installation run this code into your terminal.

pip install pynput

Suggestion : 5

If the user accepts the input on the prompt, while this coroutine is not yet finished , an asyncio.CancelledError exception will be thrown in this coroutine.,GNU readline can be configured using an .inputrc configuration file. This file contains key bindings as well as certain settings. Right now, prompt_toolkit doesn’t support .inputrc, but it should be possible in the future.,This will handle a SIGINT that was sent by an external application into the process. Handling control-c should be done by binding c-c. (The terminal input is set to raw mode, which means that a c-c won’t be translated into a SIGINT.),If the user presses only a, then nothing will happen until either a second key (like b) has been pressed or until the timeout expires (see later).

from prompt_toolkit.key_binding
import KeyBindings

bindings = KeyBindings()

def _(event):
   " Do something if 'a' has been pressed. "

def _(event):
   " Do something if Control-T has been pressed. "
stty - ixon
@bindings.add('a', 'b')
def _(event):
   " Do something if 'a' is pressed and then 'b' is pressed. "
@bindings.add('escape', 'f')
def _(event):
   " Do something if alt-f or meta-f have been pressed. "
@bindings.add('a', '<any>')
   def _(event):
from prompt_toolkit.filters
import Condition

def is_active():
   " Only activate key binding on the second half of each minute. "
return > 30

@bindings.add('c-t', filter = is_active)
def _(event):

Suggestion : 6

pyglet has support for low-level keyboard input suitable for games as well as locale- and device-independent Unicode text entry.,You can request keyboard focus for a window with the activate() method, but you should not rely on this – it may simply provide a visual cue to the user indicating that the window requires user input, without actually getting focus.,It is not possible to use pyglet’s keyboard or text events without a window; consider using Python built-in functions such as input instead.,In these cases pyglet will create a key symbol at runtime based on the hardware scancode of the key. This is guaranteed to be unique for that model of keyboard, but may not be consistent across other keyboards with the same labelled key.

def on_key_press(symbol, modifiers):

def on_key_release(symbol, modifiers):
key.ENTER or key.RETURN