How could I stop the script without having to wait for the time set for interval to pass?

Question

In this script I was looking to launch a given program and monitor it as long as the program exists. Thus, I reached the point where I got to use the threading's module Timer method for controlling a loop that writes to a file and prints out to the console a specific stat of the launched process (for this case, mspaint).

The problem arises when I'm hitting CTRL + C in the console or when I close mspaint, with the script capturing any of the 2 events only after the time defined for the interval has completely ran out. These events make the script stop.

For example, if a 20 seconds time is set for the interval, once the script has started, if at second 5 I either hit CTRL + C or close mspaint, the script will stop only after the remaining 15 seconds will have passed.

I would like for the script to stop right away when I either hit CTRL + C or close mspaint (or any other process launched through this script).

The script can be used with the following command, according to the example: python.exe mon_tool.py -p "C:\Windows\System32\mspaint.exe" -i 20

I'd really appreciate if you could come up with a working example.

I had used python 3.10.4 and psutil 5.9.0 .

This is the code:

# mon_tool.py

import psutil, sys, os, argparse
from subprocess import Popen
from threading import Timer
debug = False


def parse_args(args):   
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--path", type=str, required=True)
    parser.add_argument("-i", "--interval", type=float, required=True)
    return parser.parse_args(args)

def exceptionHandler(exception_type, exception, traceback, debug_hook=sys.excepthook):
    '''Print user friendly error messages normally, full traceback if DEBUG on.
       Adapted from http://stackoverflow.com/questions/27674602/hide-traceback-unless-a-debug-flag-is-set
    '''
    if debug:
        print('\n*** Error:')
        debug_hook(exception_type, exception, traceback)
    else:
        print("%s: %s" % (exception_type.__name__, exception))
sys.excepthook = exceptionHandler
      
def validate(data):
    try:
        if data.interval < 0:            
            raise ValueError
    except ValueError:        
        raise ValueError(f"Time has a negative value: {data.interval}. Please use a positive value")
def main():
    args = parse_args(sys.argv[1:])
    validate(args)


    # creates the "Process monitor data" folder in the "Documents" folder
    # of the current Windows profile
    default_path: str = f"{os.path.expanduser('~')}\\Documents\Process monitor data"
    if not os.path.exists(default_path):
        os.makedirs(default_path)  

    abs_path: str = f'{default_path}\data_test.txt'

    print("data_test.txt can be found in: " + default_path)


    # launches the provided process for the path argument, and
    # it checks if the process was indeed launched
    p: Popen[bytes] = Popen(args.path)
    PID = p.pid    
    isProcess: bool = True
    while isProcess:
        for proc in psutil.process_iter():
            if(proc.pid == PID):
                isProcess = False

    process_stats = psutil.Process(PID)

    # creates the data_test.txt and it erases its content
    with open(abs_path, 'w', newline='', encoding='utf-8') as testfile:
            testfile.write("")
             

    # loop for writing the handles count to data_test.txt, and
    # for printing out the handles count to the console
    def process_monitor_loop():      
        with open(abs_path, 'a', newline='', encoding='utf-8') as testfile:
            testfile.write(f"{process_stats.num_handles()}\n")
            print(process_stats.num_handles())
        Timer(args.interval, process_monitor_loop).start() 
    process_monitor_loop()
                      

if __name__ == '__main__':
    main()

Thank you!

Answer 1

You can try registering a signal handler for SIGINT, that way whenever the user presses Ctrl+C you can have a custom handler to clean all of your dependencies, like the interval, and exit gracefully. See this for a simple implementation.

Answer 2

I think you could use python-worker (link) for the alternatives

import time
from datetime import datetime
from worker import worker, enableKeyboardInterrupt

# make sure to execute this before running the worker to enable keyboard interrupt
enableKeyboardInterrupt()


# your codes
...


# block lines with periodic check
def block_next_lines(duration):
    t0 = time.time()
    while time.time() - t0 <= duration:
        time.sleep(0.05) # to reduce resource consumption


def main():
    # your codes
    ...

    @worker(keyboard_interrupt=True)
    def process_monitor_loop():
        while True:
            print("hii", datetime.now().isoformat())
            block_next_lines(3)
    
    return process_monitor_loop()

if __name__ == '__main__':
    main_worker = main()
    main_worker.wait()

here your process_monitor_loop will be able to stop even if it's not exactly 20 sec of interval

Answer 3

This is the solution for the second part of the problem, which checks if the launched process exists. If it doesn't exist, it stops the script.

This solution comes on top of the solution, for the first part of the problem, provided above by @danangjoyoo, which deals with stopping the script when CTRL + C is used.

Thank you very much once again, @danangjoyoo! :)

This is the code for the second part of the problem:

import time, psutil, sys, os
from datetime import datetime
from worker import worker, enableKeyboardInterrupt, abort_all_thread, ThreadWorkerManager
from threading import Timer

# make sure to execute this before running the worker to enable keyboard interrupt
enableKeyboardInterrupt()


# block lines with periodic check
def block_next_lines(duration):
    t0 = time.time()
    while time.time() - t0 <= duration:
        time.sleep(0.05) # to reduce resource consumption


def main():

    # launches mspaint, gets its PID and checks if it was indeed launched
    path = f"C:\Windows\System32\mspaint.exe"
    p = psutil.Popen(path)
    PID = p.pid    
    isProcess: bool = True
    while isProcess:
        for proc in psutil.process_iter():
            if(proc.pid == PID):
                isProcess = False

    interval = 5
    global counter
    counter = 0

    #allows for sub_process to run only once
    global run_sub_process_once
    run_sub_process_once = 1

    @worker(keyboard_interrupt=True)
    def process_monitor_loop():
        while True:
            print("hii", datetime.now().isoformat())


            def sub_proccess():
                '''
                Checks every second if the launched process still exists.
                If the process doesn't exist anymore, the script will be stopped.
                '''

                print("Process online:", psutil.pid_exists(PID))
                t = Timer(1, sub_proccess)
                t.start()
                global counter
                counter += 1 
                print(counter)

                # Checks if the worker thread is alive.
                # If it is not alive, it will kill the thread spawned by sub_process
                # hence, stopping the script.
                for _, key in enumerate(ThreadWorkerManager.allWorkers):
                    w = ThreadWorkerManager.allWorkers[key]
                    if not w.is_alive:
                        t.cancel()

                if not psutil.pid_exists(PID):
                    abort_all_thread()
                    t.cancel()

            global run_sub_process_once
            if run_sub_process_once:
                run_sub_process_once = 0
                sub_proccess()

            block_next_lines(interval)
    
    return process_monitor_loop()

if __name__ == '__main__':
    main_worker = main()
    main_worker.wait()

Also, I have to note that @danangjoyoo's solution comes as an alternative to signal.pause() for Windows. This only deals with CTRL + C problem part. signal.pause() works only for Unix systems. This is how it was supposed for its usage, for my case, in case it were a Unix system:

import signal, sys
from threading import Timer
 
def main():
    def signal_handler(sig, frame):
        print('\nYou pressed Ctrl+C!')
        sys.exit(0)
 
    signal.signal(signal.SIGINT, signal_handler)
    print('Press Ctrl+C')
    def process_monitor_loop():      
        try:
            print("hi")
        except KeyboardInterrupt:    
            signal.pause()
        Timer(10, process_monitor_loop).start() 
    process_monitor_loop()
 
 
if __name__ == '__main__':
    main()

The code above is based on this.