TheMaxai
@TheMaxai

Как сделать закрытие консоли если будет нажатие на кнопку?

Суть в том что это цикл и input() нельзя применять, т.к. останавливать программу нельзя.
нужно что-то типа input() но в отдельном потоке...
но в multiprocessing применять input нельзя..
  • Вопрос задан
  • 387 просмотров
Решения вопроса 1
sergey-gornostaev
@sergey-gornostaev Куратор тега Python
Седой и строгий
Многопоточный вариант
import time
from threading import Thread

run = True

def ticker():
    while run:
        time.sleep(1)
        print('Тик')
        time.sleep(1)
        print('Так')


# Запуск функции ticker в отдельном потоке.
# Параметр daemon=True нужен чтобы
# дочерний поток умирал вместе с основным
# в случае внештатного выхода.
Thread(target=ticker, daemon=True).start()

while run:
    command = input('Для выхода введите "exit"\n')
    if command.lower() == 'exit':
        run = False

Асинхронный вариант
import asyncio
import signal

def shutdown():
    # Отменяем все задачи, кроме вызвавшей
    for task in asyncio.Task.all_tasks():
        if task is not asyncio.tasks.Task.current_task():
            task.cancel()


async def user_io():
    loop = asyncio.get_event_loop()
    # Ждём действия от пользователя
    while True:
        # Запускаем input() в отдельном потоке и ждём его завершения
        command = await loop.run_in_executor(None, input, 'Для выхода введите "exit"\n')
        if command.lower() == 'exit':
            shutdown() # Отменяем все задачи
            break      # и выходим из цикла


# Сопрограмма, выполняемая параллельно с ожиданием пользовательского ввода
async def task_manager():
    counter = 0
    while True:
        try:
            await asyncio.sleep(1)
        except asyncio.CancelledError:
            break # Выходим из цикла, если задачу отменили
        counter += 1
        print("I'm a task manager {}!".format(counter))


if __name__ == '__main__':
    # Устанавливаем обработчик Ctrl+C
    signal.signal(signal.SIGINT, lambda n, f: shutdown())

    # Запускаем цикл событий
    loop = asyncio.get_event_loop()
    # Задача ждущая завершения сопрограм user_io и task_manager
    main_task = asyncio.wait([user_io(), task_manager()])
    try:
        loop.run_until_complete(main_task)
    except asyncio.CancelledError:
        # Позволяем main_task завершиться
        loop.run_until_complete(main_task)
    loop.close()
Ответ написан
Пригласить эксперта
Ваш ответ на вопрос

Войдите, чтобы написать ответ

Войти через центр авторизации
Похожие вопросы