Как немедленно прервать выполнение указанной функции при нажатии на кнопку?

Question

[SoftHardcore]

Нужно, чтобы после нажатия на клавишу "s" или кнопку STOP, выполнение функции mainProc() тут же прерывалось. То есть, например, чтобы эта функция, которая "рисует" курсором квадрат, тут же переставала водить курсором, если в этот момент была нажата кнопка.

import time
import keyboard
import pyautogui
import threading
from tkinter import *


root = Tk()
root.title("Label")


time.sleep(0.5)


stop = True


def button_stop_command():
    global stop
    stop = True


def button_start_command():
    global stop
    if stop == True:
        stop = False
        while stop == False:
            mainProc()


def button_starter():
    t = threading.Thread(target=button_start_command)
    t.start()


button_start = Button(root, text="START", padx=30, pady=20, command=button_starter)
button_start.grid(columnspan=1, row=1,column=0)

button_stop = Button(root, text="STOP", padx=32, pady=20, command=button_stop_command)
button_stop.grid(row=2, column=0)


keyboard.add_hotkey('a', button_starter)
keyboard.add_hotkey('s', button_stop_command)


def mainProc():
    pyautogui.move(150, 0, 0.5, pyautogui.easeOutQuad)
    pyautogui.move(0, 150, 0.5, pyautogui.easeOutQuad)
    pyautogui.move(-150, 0, 0.5, pyautogui.easeOutQuad)
    pyautogui.move(0, -150, 0.5, pyautogui.easeOutQuad)


root.mainloop()

Уточняю, что необходимо именно прервать сразу при нажатии кнопки. Вот пример кода, который при нажатии кнопки "d" способен прервать программу, из-за чего движение курсора тут же прекращается. Однако это решение завершает работу всей программы, а нужно прервать выполнение только функции, ну или конкретного потока.

Посмотреть пример

stop = True
exit = False


def buttonStartCommand():
    global stop
    if stop == True:
        stop = False
        while stop == False:
            mainProc()


def buttonStopCommand():
    global stop
    stop = True


def buttonExitCommand():
    global exit
    exit = True


def buttonStarter():
    t = threading.Thread(target=buttonStartCommand)
    t.start()


keyboard.add_hotkey('a', buttonStarter)
keyboard.add_hotkey('s', buttonStopCommand)
keyboard.add_hotkey('d', buttonExitCommand)


def mainProc():
    pyautogui.move(150, 0, 0.5, pyautogui.easeOutQuad)
    pyautogui.move(0, 150, 0.5, pyautogui.easeOutQuad)
    pyautogui.move(-150, 0, 0.5, pyautogui.easeOutQuad)
    pyautogui.move(0, -150, 0.5, pyautogui.easeOutQuad)


while not exit:
    time.sleep(0.04)

Answer 1

[Hemul GM]

Он и прервется, когда из функции выйдет
Хочешь, чтоб между смещениями мыши прервалось, делай после каждого вызова проверку

Comments

[SoftHardcore]

Нет, я хочу, чтобы прервалось сразу после нажатия кнопки, то есть и во время смещения курсора тоже должно прерываться. Кроме того, такой способ будет несколько ухудшать читаемость и заставляет прописывать проверку фактически после каждой строчки кода.

[Hemul GM]

SoftHardcore, больше никак не сделать. Это синхронный код. Сделай метод свой на смещение и там проверку сделай. И будет тебе опять 4 строчки

[SoftHardcore]

Hemul GM, понятно. Тогда насчет метода на смещение: я не знаю что это, вы можете помочь и показать как это? Вообще, в реальной примере у меня функция mainProc() не только курсором двигает, но и щелкает кнопки мыши и выполняет более сложные методы, например может вызывать самописную функцию с drag and drop. И еще, вы упомянули, что код синхронный, а я ведь вроде многопоточность в него засунул. Может тогда можно резко остановить выполнение потока, в котором будет функция mainProc()?

[Hemul GM]

SoftHardcore, у тебя вызывается 4 раза функция pyautogui.move. Сделай метод, который принимает параметры, которые ты в ней указываешь и перед вызовом pyautogui.move сделай проверку

def move(x, y...):
  If not stop:
    pyautogui.move(x, y ..)

И вызывай в mainProc этот метод

Answer 2

[Wearant]

def button_start_command():
    global stop
    if stop == True:
        stop = False
        mainProc()

def mainProc():
    while 1:
        pyautogui.move(150, 0, 0.5, pyautogui.easeOutQuad)
        if stop == True:
            break
        pyautogui.move(0, 150, 0.5, pyautogui.easeOutQuad)
        if stop == True:
            break
        pyautogui.move(-150, 0, 0.5, pyautogui.easeOutQuad)
        if stop == True:
            break
        pyautogui.move(0, -150, 0.5, pyautogui.easeOutQuad)
        if stop == True:
            break

Comments

[SoftHardcore]

Этот вариант имеет две проблемы:
1) программа каждый раз дожидается конца выполнения метода move() и не способна прервать его выполнение сразу после нажатия кнопки;
2) этот способ заметно ухудшает читаемость кода и при этом заставляет прописывать проверку после каждого шага.

Answer 3

[Dmitrii]

Если вы хотите прерывать код по какому-то событию практически в произвольном месте (в любой момент), то тогда вам необходим объект класса, который будет определять каким функциям сейчас работать (своеобразный scheduler), то есть работать ли основной программе или же работать обработчику нажатия кнопки, движения мыши. В данном случае имеет смысл говорить о вытесняющей многозадачности.

Если же вы не хотите строить сложную кастомную систему, то можете написать, например, асинхронный код (поскольку в основе уже есть определенный scheduler) и реализовать свою логику вытеснения одних задач другими.

Comments

[SoftHardcore]

Первый вариант выглядит сложным, но и как реализовать второй вариант тоже не понятно. По сути сейчас, при выполнении цикла по рисованию курсором квадрата, программа способна зарегистрировать, что нажата кнопка, и в таком случае по сути прерывает цикл, но прерывает его только после окончания текущего витка. Если бы существовал какой-то метод, типа "break mainProc()", который можно было бы вызвать из другой функции в рамках асинхронной работы обеих функций, и таким образом через одну функцию прерывать работу другой, то тогда асинхронность мне бы помогла. Однако ни о каких-либо способов прямого контроля за выполнением функции или потока с возможностью его экстренного прерывания по кнопке я не нашел.

[Dmitrii]

SoftHardcore, да, все так. В вашем случае, можно также создать два потока и в случае необходимости передавать данные (например, через очередь) из одного в другой (что-то вроде команд). Единственный вариант дополнительный для вас для получения быстрого отклика на клик пользователя - разбить выполнение длительной функции (например, рисования целого объекта) на части и тогда у вас получится проверять блокировку значительно чаще, получить меньшее время отклика на событие

[SoftHardcore]

Dmitrii, а вы можете дать пример кода? А то так на словах вообще не понятно, как такое реализовать. Я не нашел никаких методов, чтобы контролировать работу функций или потоков и прерывать их "на горячую" в любой момент при нажатии кнопки.

[Dmitrii]

SoftHardcore, пример того как остановить функцию на "горячую" не выйдет, кроме аварийного завершения потока, но, думаю, вам не это нужно. Самое подходящее решение, на мой взгляд - вы можете разбить программу на очень маленькие функции, между которыми можно проверять было ли события (нажатие кнопки) или нет. Это можно делать даже в одном потоке, но тогда вы можете пропускать события, которые успели начаться и завершиться пока функция выполнялась (если она все же получилась длительная).
Вот пример для двух потоков и mutex'a. Можно также использовать event, вместо mutex.
Также тут есть недостаток, что первый поток может заблокировать mutex, и второй останется в вечном ожидании мьютекса. Это можно исправить если запихать checkEvent в класс, который в своем деструкторе будет разблокировать mutex. Или просто в конце функции проверять, что mutex точно свободен. Также есть проблема многопоточности в python, связаная с GIL, если интересно почитайте.

import threading
import time
import random
import string

def checkEvent(mutex):
    for i in range(1000):
        if mutex.locked():
            mutex.release()
            print("Mutex released", "=" * 30)
        else:
            mutex.acquire()
            print("Mutex locked", "=" * 30)
        time.sleep(0.5)

def longFunc(string, mutex):
    while len(string) > 10:
        print(string[:10])
        string = string[10:]
        time.sleep(0.5)
        while mutex.locked():
            time.sleep(0.1) # or do whatever you need
    else:
        print(string)

if __name__ == "__main__":

    s = "".join(random.choices(string.ascii_uppercase + string.digits, k=1000))

    mutex = threading.Lock()
    t1 = threading.Thread(target = longFunc, args = (s, mutex))
    t2 = threading.Thread(target = checkEvent, args = (mutex, ))
    t1.start()
    t2.start()

    t1.join()
    t2.join()

[SoftHardcore]

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

import threading
import time
import random
import string
import pyautogui
import keyboard

stop = False

def checkEvent(mutex):
    while True:
        if mutex.locked():
            mutex.release()
            print("Mutex released", "=" * 30)
        else:
            mutex.acquire()
            print("Mutex locked", "=" * 30)
        time.sleep(0.5)

def longFunc(string, mutex):
    while len(string) > 10: # ℳагия, без которой почему-то не будет работать
        time.sleep(0.5)
        print('Начинаем выполнение основной функции mainProc')
        mainProc()
        while mutex.locked():
            time.sleep(0.1) # or do whatever you need
            print('Поток заблокирован!')
    else:
        print(string, 'Если видишь этот текст, значит магия со string закончилась и что-то пошло не так.')

def mainProc():
    pyautogui_move(150, 0, 0.5, pyautogui.easeOutQuad)
    pyautogui_move(0, 150, 0.5, pyautogui.easeOutQuad)
    pyautogui_move(-150, 0, 0.5, pyautogui.easeOutQuad)
    pyautogui_move(0, -150, 0.5, pyautogui.easeOutQuad)

def pyautogui_move(*args):
    if stop == False: # унылая проверка... и в реальном коде такие проверки
                      # придется прописывать чуть ли не для каждой функции,
                      # и не дай бог, если придется что-то менять в логике,
                      # ибо менять эту строчку придется для каждой подобной функции,
                      # а их могут быть сотни;
                      # по идее эта проверка должна как-то автоматически прописываться
                      # после каждой команды внутри функции mainProc(), если это осуществляемо
        pyautogui.move(*args)

def button_start_command():
    print("СТАРТ")
    global stop
    stop = False

def button_stop_command():
    print("ОСТАНОВКА")
    global stop
    stop = True

if __name__ == "__main__":

    s = "".join(random.choices(string.ascii_uppercase + string.digits, k=1000))

    mutex = threading.Lock()
    t1 = threading.Thread(target = longFunc, args = (s, mutex))
    t2 = threading.Thread(target = checkEvent, args = (mutex, ))
    t1.start()
    t2.start()

    keyboard.add_hotkey('a', button_start_command)
    keyboard.add_hotkey('s', button_stop_command)

    t1.join()
    t2.join()

[Dmitrii]

SoftHardcore, string в моем случае было иллюстрацией длительной задачи - печать, которую я выполнял по чуть-чуть проверяя флаг. Естественно, это не код для коммерческого решения. Вынесеите интерфейс взаимодействия в отдельный класс, метод которого вы будете вызывать уже в нужном месте (например, button_stop_command).
Из каждой функции, конечно же, нужно вынести проверки и проверять до их выполнения. Воспользуйтесь паттерном цепочка команд, итератор.