Python3: обнаружение нажатий клавиш асинхронно и передача их в основной поток - PullRequest
Новая
       29

Python3: обнаружение нажатий клавиш асинхронно и передача их в основной поток

1 голос
/ 11 июня 2019

Отказ от ответственности : операторы импорта находятся внутри функций, я знаю, что это необычно. Я показываю всю свою программу здесь, показывая ее функцию для функции, рассказывая о своей проблеме и своем мышлении. В действительности я делаю что-то другое, я сделал этот минимальный пример только для этого вопроса Stackoverflow. Есть дублирующие вопросы, но я не нашел в них хороших ответов, поскольку они говорят только «использовать многопоточность» (например, этот ответ ). Этот конкретный вопрос касается того, как использовать многопоточность.

История : я запускаю программу на Python. Допустим, это цикл времени, продолжающийся до бесконечности. Это просто бежит счастливо. Например, 

def job(threadname, q):
  from time import sleep
  c = 0
  while True:
    sleep(0.1) #slow the loop down 
    c += 1
    print(c)

Что я хочу сделать, так это то, что он асинхронно обнаруживает нажатие клавиши на stdin, а затем прерывает выполнение, так что я могу делать все, что захочу, в функции, в которой оно прервано (или если я запускаю его с python3 -i program.py, чтобы переключиться на REPL со всеми загруженными моими модулями, помните, что это минимальный пример, в котором я не хочу особо подчеркивать такие проблемы).

Моя идея была: у меня есть одна функция, которая асинхронно получает нажатие клавиши, отправляет ее через очередь в другой поток, и она работает. Поэтому я расширил функцию задания так: 

def job(threadname, q):
  from time import sleep
  c = 0
  while True:
    sleep(0.1) #slow the loop down 
    c += 1
    print(c)
    ch = q.get() #extension for multithreading
    handle_keypress(ch) #extension for handling keypresses

Код для handle_keypress(ch): 

def handle_keypress(key):
  if (key == "q"):
    print("Quit thread")
    exit(0)
  elif (key == "s"):
    print("would you like to change the step size? This has not been implemented yet.")
  else:
    print("you pressed another key, how nice! Unfortunately, there are not anymore options available yet.")

Другими словами, это не так интересно, кроме как продемонстрировать, что я хочу это сделать.

Сначала проблема казалась в функции job(). Виновником является q.get(), который висит. Тем не менее, он зависает, потому что мой входной поток по какой-то причине не асинхронный и блокирует. Понятия не имею, как его разблокировать.

Это функция моего потока ввода: 

def get_input(threadname, q):
  #get one character, this code is adapted from /484486/python-chitaet-odin-simvol-ot-polzovatelya
  while True:
    import sys, tty, termios
    fd = sys.stdin.fileno()
    old_settings = termios.tcgetattr(fd)
    try:
        tty.setraw(sys.stdin.fileno())
        ch = sys.stdin.read(1)
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
    q.put(ch)

Для меня очевидно, что sys.stdin.read(1) блокирует, но я не знаю, как его разблокировать. В таком состоянии, как сейчас, я не могу придумать, как справиться с ситуацией с отравленной таблеткой, поэтому q.get() в функции job() блокирует.

Я запускаю программу, вызывая следующую функцию: 

def run_program():
  from threading import Thread
  from queue import Queue
  queue = Queue()
  thread1 = Thread( target=get_input, args=("Thread-1", queue) )
  thread2 = Thread( target=job, args=("Thread-2", queue) )

  thread1.start()
  thread2.start()
  thread1.join()
  thread2.join()

Мои вопросы : так вы бы разработали программу для работы с асинхронными нажатиями клавиш? Если да, то как разблокировать функцию get_input()?

1 Ответ

0 голосов
/ 11 июня 2019

Благодаря Sav я нашел способ ответить на этот вопрос.На мой взгляд, его комментарий является ответом.Так что, если он перепишет свой комментарий.Я приму его ответ.Сейчас я покажу, какую часть кода я изменил, чтобы заставить работать неблокирующую реализацию: 

def job(threadname, q):
  from queue import Empty
  from time import sleep
  c = 0
  while True:
    sleep(0.1) #slow the loop down 
    c += 1
    print(c)
    #Below is the changed part
    ch = None
    try:
      ch = q.get(block=False)
    except Empty:
      pass
    if ch is not None:
      handle_keypress(ch)
...