Python: поймать команду Ctrl-C. Подскажите, действительно хотите выйти (y/n), возобновите выполнение, если нет

У меня есть программа, которая может иметь длительное выполнение. В основном модуле у меня есть следующее:

import signal
def run_program()
   ...time consuming execution...

def Exit_gracefully(signal, frame):
    ... log exiting information ...
    ... close any open files ...
    sys.exit(0)

if __name__ == '__main__':
    signal.signal(signal.SIGINT, Exit_gracefully)
    run_program()

Это работает нормально, но мне бы хотелось иметь возможность приостанавливать выполнение после перехвата SIGINT, подсказывая пользователю, действительно ли он хочет выйти, и возобновляя его с того места, где я остановился в run_program(), если он решит, что не хочет выходить.

Единственный способ, который я могу придумать, - это запустить программу в отдельном потоке, заставив основной поток ожидать ее и быть готовым поймать SIGINT. Если пользователь хочет выйти из основного потока, он может выполнить очистку и убить дочерний поток.

Есть ли более простой способ?


person Colin M    schedule 07.08.2013    source источник
comment
Что не так с if raw_input("Really quit? y/n").lower().startswith('y'): кодом выхода?   -  person user2357112 supports Monica    schedule 08.08.2013


Ответы (3)


Обработчики сигналов Python не кажутся настоящими обработчиками сигналов; то есть они происходят постфактум, в обычном потоке и после того, как обработчик C уже вернулся. Таким образом, вы попытаетесь поместить свою логику выхода в обработчик сигнала. Поскольку обработчик сигнала работает в основном потоке, он также блокирует выполнение там.

Что-то вроде этого, кажется, хорошо работает.

import signal
import time
import sys

def run_program():
    while True:
        time.sleep(1)
        print("a")

def exit_gracefully(signum, frame):
    # restore the original signal handler as otherwise evil things will happen
    # in raw_input when CTRL+C is pressed, and our signal handler is not re-entrant
    signal.signal(signal.SIGINT, original_sigint)

    try:
        if raw_input("\nReally quit? (y/n)> ").lower().startswith('y'):
            sys.exit(1)

    except KeyboardInterrupt:
        print("Ok ok, quitting")
        sys.exit(1)

    # restore the exit gracefully handler here    
    signal.signal(signal.SIGINT, exit_gracefully)

if __name__ == '__main__':
    # store the original SIGINT handler
    original_sigint = signal.getsignal(signal.SIGINT)
    signal.signal(signal.SIGINT, exit_gracefully)
    run_program()

Код восстанавливает исходный обработчик сигнала на время raw_input; Сам raw_input нельзя повторно ввести, и повторный вход в него приведет к тому, что RuntimeError: can't re-enter readline будет поднят из time.sleep, чего мы не хотим, поскольку его труднее поймать, чем KeyboardInterrupt. Вместо этого мы позволяем 2 последовательным нажатиям Ctrl-C поднять KeyboardInterrupt.

person Antti Haapala    schedule 07.08.2013
comment
Ух ты! Это действительно круто. Тем не менее, я думаю, что возможность crtl-c из подсказки должна быть в декораторе, так как это делает код гораздо менее загадочным. Было бы уместно предоставить этот способ в качестве ответа (поскольку я не могу отредактировать ваш ответ, чтобы добавить этот альтернативный способ)? - person mr2ert; 08.08.2013
comment
Нет, декоратор сделал бы это более загадочным, а настройка обработчика сигнала на самом деле является частью логики функции. Может быть, я просто добавлю несколько комментариев :D - person Antti Haapala; 08.08.2013
comment
Декораторы немного волшебны... Я сохраню свою версию декоратора в модуле, который я пишу для этого (потому что он такой милый :)). Однако я хотел бы отметить, что вы можете сделать это, не меняя обработчик сигнала, и выйти, только если пользователь вводит y в приглашение. - person mr2ert; 08.08.2013
comment
Круто, это работает. У меня есть проблема с запуском вашего кода Antti, где выход из обработчика сигнала вызывает IOERROR: Прерванный вызов функции. Только когда я устанавливаю сон на 0,001 секунды, он работает, он ломается на 0,01 или выше. Насколько я могу судить, это также проблема только в Windows, код работает нормально, как и в Cygwin. - person Colin M; 08.08.2013
comment
Под выходом я не подразумеваю вызов sys.exit(). Просто осознал потенциальную двусмысленность. По-видимому, ему не нравится пытаться возобновить вызов sleep() после возврата из обработчика сигнала. - person Colin M; 08.08.2013
comment
А, Винда. Windows даже не имеет сигналов, так как можно предположить, что эмуляция работает идеально. Кажется, что если CTRL-C перехватывается обработчиком сигнала, то time.sleep выдает IOError? Уведомление. также документировано, что time.sleep может вернуться раньше, если сигнал будет доставлен (возможно, вызовет исключение) - person Antti Haapala; 09.08.2013
comment
Стратегия не работает, если у нас многопоточное приложение. - person Ciasto piekarz; 21.06.2014
comment
зачем вам хранить оригинальный обработчик SIGINT? плюс я не думаю, что это изящный выход, я бы скорее назвал его некрасивым выходом !! - person Ciasto piekarz; 21.06.2014
comment
Обработчики сигналов @san bc не подлежат повторному входу, и я хочу запустить исходный обработчик, если снова нажать ctrl-c. - person Antti Haapala; 22.06.2014
comment
@san также, сигнал должен в основном работать в многопоточном приложении, в котором 1 основной поток обрабатывает все входные данные, но обработчик сигнала должен быть установлен в этом основном потоке... - person Antti Haapala; 22.06.2014
comment
@AnttiHaapala - что именно произойдет, если обработчик сигнала не будет повторно входить? т.е. если строка signal.signal(signal.SIGINT, original_sigint) функции exit_gracefully была закомментирована? - person Matteo; 18.08.2016
comment
@AnttiHaapala - Кроме того, можно ли изменить эту программу, чтобы полностью игнорировать обработчики сигналов? Таким образом, что только убийство процесса остановит программу? - person Matteo; 18.08.2016
comment
@Matteo, обработчики сигналов posix, зарегистрированные signal.signal, являются одноразовыми, поэтому, если они не были перерегистрированы, следующий ctrl-c выдаст Keyboardinterrupt - person Antti Haapala; 18.08.2016
comment
Есть несколько способов игнорировать ctrl-c в POSIX: вы можете заставить терминал игнорировать ctrl-c, вы можете сделать так, чтобы программа не имела управляющего терминала, или вы можете игнорировать сигнал; поэтому вы должны просто использовать signal.signal(signal.SIGINT, signal.SIG_IGN), чтобы полностью заблокировать ctrl-c. - person Antti Haapala; 18.08.2016
comment
Что плохого может произойти в raw_input и почему недостаточно поймать KeyboardInterrupt? Почему важно восстановить исходный обработчик sigint при обработке sigint в exit_gracefully? - person so.very.tired; 03.09.2016
comment
@so.very.tired ответила^ - person Antti Haapala; 03.09.2016

из https://gist.github.com/rtfpessoa/e3b1fe0bbfcd8ac853bf

#!/usr/bin/env python

import signal
import sys

def signal_handler(signal, frame):
  # your code here
  sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)

Пока!

person Marc    schedule 04.09.2019
comment
хорошо, извини... сейчас у меня нет времени, как только смогу, я последую твоему совету, спасибо - person Marc; 05.09.2019
comment
Не прерывает вызов time.sleep() для меня в Python 3.7.3. - person Ben Slade; 27.10.2020

когда процедура закончится, сделайте что-нибудь

предположим, вы просто хотите, чтобы процедура что-то делала после завершения задачи

import time

class TestTask:
    def __init__(self, msg: str):
        self.msg = msg

    def __enter__(self):
        print(f'Task Start!:{self.msg}')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('Task End!')

    @staticmethod
    def do_something():
        try:
            time.sleep(5)
        except:
            pass

with TestTask('Hello World') as task:
    task.do_something()

когда процесс покидает with, который будет работать __exit__, даже если KeyboardInterrupt будет одинаковым.

если вам не нравится видеть ошибку, добавьте try ... except ...

@staticmethod
def do_something():
    try:
        time.sleep(5)
    except:
        pass

пауза, продолжение, сброс и т. д.

У меня нет идеального решения, но оно может оказаться полезным для вас.

Это означает разделить ваш процесс на множество подпроцессов и сохранить его завершенным. Он не будет выполняться снова, так как вы обнаружите, что он уже выполнен.

import time
from enum import Enum

class Action(Enum):
    EXIT = 0
    CONTINUE = 1
    RESET = 2

class TestTask:
    def __init__(self, msg: str):
        self.msg = msg

    def __enter__(self):
        print(f'Task Start!:{self.msg}')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print('Task End!')

    def do_something(self):
        tuple_job = (self._foo, self._bar)  # implement by yourself
        list_job_state = [0] * len(tuple_job)
        dict_keep = {}  # If there is a need to communicate between jobs, and you don’t want to use class members, you can use this method.
        while 1:
            try:
                for idx, cur_process in enumerate(tuple_job):
                    if not list_job_state[idx]:
                        cur_process(dict_keep)
                        list_job_state[idx] = True
                if all(list_job_state):
                    print('100%')
                    break
            except KeyboardInterrupt:
                print('KeyboardInterrupt. input action:')
                msg = '\n\t'.join([f"{action + ':':<10}{str(act_number)}" for act_number, action in
                                   enumerate([name for name in vars(Action) if not name.startswith('_')])
                                   ])
                case = Action(int(input(f'\t{msg}\n:')))
                if case == Action.EXIT:
                    break
                if case == Action.RESET:
                    list_job_state = [0] * len(tuple_job)

    @staticmethod
    def _foo(keep_dict: dict) -> bool:  # implement by yourself
        time.sleep(2)
        print('1%')
        print('2%')
        print('...')
        print('60%')
        keep_dict['status_1'] = 'status_1'
        return True

    @staticmethod
    def _bar(keep_dict: dict) -> bool:  # implement by yourself
        time.sleep(2)
        print('61%')
        print(keep_dict.get('status_1'))
        print('...')
        print('99%')
        return True

with TestTask('Hello World') as task:
    task.do_something()

консоль

input action number:2
Task Start!:Hello World
1%
2%
...
60%
KeyboardInterrupt. input action:
        EXIT:     0
        CONTINUE: 1
        RESET:    2
:1
61%
status_1
...
99%
100%
Task End!

person Carson    schedule 23.09.2019