ich habe ein lange dauerndes Programm geschrieben, das u.a. Screenshots macht. Deshalb wäre es wichtig, dass das Programm stoppt, falls das niedriger-Akkustand-Fenster kommt:
Code: Alles auswählen
# 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()Code: Alles auswählen
# 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()
