Основная программа не останавливается, несмотря на сигнал остановки. ⇐ Python

[Anonymous]

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

Код: Выделить всё

# main_script.py

import time
import threading
from queue import Queue, Empty
from threading import Event
import random
from concurrent.futures import ThreadPoolExecutor
from battery_monitor import monitor_low_battery_window, stop_event

directory = "C:/Users/User/pictures"

def process_files_with_work_stealing(
main_queue: Queue,
steal_queue: Queue,
stop_signal: Event
) -> None:
"""Processes files using a work-stealing approach, distributing tasks between queues."""
while not stop_signal.is_set():
task = None

try:
task = main_queue.get_nowait()
except Empty:
# If the main queue is empty, try to steal a task from the other queue
try:
task = steal_queue.get_nowait()
except Empty:
# Both queues are empty, briefly wait before retrying
time.sleep(0.1)
continue

# Process file...
pass

def main() -> None:
queue_a = Queue()
queue_b = Queue()
stop_signal = Event()

try:
with ThreadPoolExecutor(max_workers=2) as executor:
for folder in directory.iterdir():
if folder.is_dir():
for file in folder.glob("*.png"):
if file.is_file():
if random.random() < 0.5:
queue_a.put(file)
else:
queue_b.put(file)

executor.submit(process_files_with_work_stealing, queue_a, queue_b, stop_signal)
executor.submit(process_files_with_work_stealing, queue_b, queue_a, stop_signal)

# Wait until all files are processed (queues are empty)
while not queue_a.empty() or not queue_b.empty():
time.sleep(1)

stop_signal.set()
except KeyboardInterrupt:
print("\nProgram interrupted by the user.")
finally:
stop_event.set()  # Signals the monitor thread to exit
monitor_thread.join()  # Waits for the thread to be closed
print("Main thread completed. Monitor thread joined.")

if __name__ == '__main__':
monitor_thread = threading.Thread(target=monitor_low_battery_window)
monitor_thread.start()

try:
main()
except KeyboardInterrupt:
print("\nMain script interrupted and stopped.")
stop_event.set()  # Stop the monitoring thread
monitor_thread.join()

Поэтому важно, чтобы эта основная программа останавливалась, когда вспомогательная программа, работающая параллельно, обнаруживает всплывающее окно о низком заряде батареи:

Код: Выделить всё

# battery_monitor.py

import time
import pyautogui
import cv2
import numpy as np
from colored_print import Print
import threading

stop_event = threading.Event()

def monitor_low_battery_window() -> None:
"""Monitors the screen for a low battery warning and signals the program to exit."""
print("Monitor thread started.")
while not stop_event.is_set():
if check_low_battery_window(region=(900, 720, 720, 200)):  # (left, top, width, height)
Print.critical("Low battery warning detected.  Exiting program...")
stop_event.set()
break
time.sleep(1)
print("Monitor thread exiting.")

def check_low_battery_window(region: tuple) -> bool:
"""Checks for a low battery warning window in the specified region of the screen."""
screenshot = pyautogui.screenshot()
screenshot = cv2.cvtColor(np.array(screenshot), cv2.COLOR_RGB2BGR)

if region:
left, top, width, height = region
screenshot = screenshot[top:top + height, left:left + width]

template = cv2.imread("low_battery_window.png")
if template is None:
Print.error("Low battery template image not found.")
return False

result = cv2.matchTemplate(screenshot, template, cv2.TM_CCOEFF_NORMED)
threshold = 0.8
locations = np.where(result >= threshold)

return locations[0].size > 0

if __name__ == '__main__':
try:
print("Starting battery monitor test...")
monitor_thread = threading.Thread(target=monitor_low_battery_window)
monitor_thread.start()

while not stop_event.is_set():
time.sleep(0.5)

print("Test completed. Stopping monitor thread.")

monitor_thread.join()
print("Monitor thread stopped successfully.")
except KeyboardInterrupt:
print("\nTest interrupted by user.")
stop_event.set()
monitor_thread.join()

Когда я запускаю только сценарий Battery_monitor.py, все работает так, как хотелось. Однако основная программа не останавливается ни тогда, когда Battery_monitor.py обнаруживает окно с низким зарядом батареи, ни когда я запускаю прерывание клавиатуры.

Подробнее здесь: https://stackoverflow.com/questions/792 ... top-signal