在這個腳本中，我希望啟動一個給定的程式并在程式存在時對其進行監控。因此，我達到了使用執行緒模塊 Timer 方法來控制一個回圈的地步，該回圈寫入檔案并將啟動行程的特定統計資訊列印到控制臺（在本例中為 mspaint）。
當我在控制臺中按 CTRL C 或關閉 mspaint 時出現問題，腳本僅在為間隔定義的時間完全用完后才捕獲 2 個事件中的任何一個。這些事件使腳本停止。
例如，如果為間隔設定了 20 秒的時間，一旦腳本啟動，如果在第 5 秒我按 CTRL C 或關閉 mspaint，則腳本將僅在剩余的 15 秒過去后停止。
當我按下 CTRL C 或關閉 mspaint（或通過此腳本啟動的任何其他行程）時，我希望腳本立即停止。
根據示例，該腳本可以與以下命令一起使用：python.exe mon_tool.py -p "C:\Windows\System32\mspaint.exe" -i 20
如果您能提出一個可行的示例，我將不勝感激。
我用過 python 3.10.4 和 psutil 5.9.0 。
這是代碼：
# 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()
謝謝！
uj5u.com熱心網友回復：
我認為您可以使用python-worker（鏈接）作為替代品
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()
process_monitor_loop即使間隔不完全是 20 秒，您也可以在這里停下來
uj5u.com熱心網友回復：
您可以嘗試為 SIGINT 注冊信號處理程式，這樣每當用戶按下 Ctrl C 時，您都可以使用自定義處理程式來清除所有依賴項，例如間隔，然后優雅地退出。請參閱此以獲得簡單的實作。
uj5u.com熱心網友回復：
這是問題第二部分的解決方案，它檢查啟動的行程是否存在。如果它不存在，它將停止腳本。
該解決方案是解決問題的第一部分，由@danangjoyoo提供，它處理在使用 CTRL C 時停止腳本。
再次非常感謝你，@danangjoyoo！:)
這是問題第二部分的代碼：
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()
另外，我必須注意@danangjoyoo的解決方案是 Windows 的 signal.pause() 的替代方案。這僅處理 CTRL C 問題部分。signal.pause() 僅適用于 Unix 系統。這就是它的使用方式，就我而言，如果它是一個 Unix 系統：
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()
上面的代碼就是基于這個。
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/486663.html
標籤：Python python-3.x 视窗 多线程 计时器