【Python】class Aから他クラスのclass Bで実行中のプログラム処理を中断したい

実現したいこと

前提

＜添付プログラムの概要＆やりたいこと＞
１：処理を実行するカウントダウンの表示「中断ボタン → class Interruption()によって中断後の案内文が出る」
２：class Timeの処理が実行される「中断ボタン → class Interruption()によって中断後の案内文が出る」
３：処理終了後に画面を閉じるカウントダウンの表示
４：全プロセス終了

＜＜＜【　質問理由・目的　】＞＞＞

・class GUIのdef processingにおいて、中断ボタンが表示されます。そして、def processingでは他クラスであるclass Time()から呼び出し実行した2つの関数が処理されますが、処理中の任意のタイミングで中断ボタンを押すことでそれらの処理を中断したいです＜上記２：の中断～に該当＞
※仮にdef Time_sleepの処理が1日だった場合、やりたい処理は下記の通り
×「中断ボタン → def Time_sleepの処理後に中断：長い処理を中断させたいのに終わってからでは意味がないため」
〇「中断ボタン → def Time_sleepの処理中での中断し、Time_arg = Time()の処理自体を終了して次の処理を実行」

発生している問題・エラーメッセージ

＜添付プログラムでの問題＞：

①：def processingの時点でそもそも中断ボタンが押せなくなっている

②：問題①はthreading関数を用いて別スレッドでclass Time()を実行すれば中断ボタンを押せるようにはなる。しかし、class GUI上でのプロセスのみが無事に終了し、別スレッドであるclass Time()の関数たちは中断されず処理される。

該当のソースコード

Python(jupyter)
1import os
2import sys
3import time
4import tkinter as tk
5
6
7class Time:
8    def __init__(self):
9        self.Time_sleep()
10        self.Num_calc()
11
12    def Time_sleep(self):
13        time.sleep(100)  # 100s 待機
14
15    def Num_calc(self):
16        for i in range(250000000):  # 約5sの処理時間
17            pass
18
19
20class GUI:
21    def __init__(self):  # 外枠となる設計図を作成
22        self.count = 3
23
24        self.root = tk.Tk()
25        self.root.geometry("1400x200")
26        self.integer_variable = tk.BooleanVar()
27        self.label = tk.Label(self.root, text="")
28        self.label.pack()
29        self.button = tk.Button(self.root, text="中断", command=self.end)
30        self.button.pack()
31        self.label.after(1, lambda: self.countdown(self.count, self.processing, True))
32
33    def countdown(self, remaining, callback, coll):
34        if coll == False:
35            puo = "\n処理が完了しました"  # \nは表示を真ん中にするための調整
36            pip = "閉じます"
37
38            self.button.pack_forget()
39        else:
40            puo = "処理を実行します"
41            pip = "開始"
42
43        if remaining <= 0:
44            callback()
45        else:
46            self.root.title("Backup")
47            self.label.configure(
48                text="{}：{}秒後に{}".format(puo, remaining, pip), font=("Helvetica", 20)
49            )
50            self.root.after(1000, lambda: self.countdown(remaining - 1, callback, coll))
51
52    def processing(self):
53        # ラベルを更新
54        self.label.configure(text="処理中です", font=("Helvetica", 20))
55        self.root.update_idletasks()
56
57        Time_arg = Time()
58
59        self.countdown(5, self.end_taken, False)
60
61    def end(self):
62        self.integer_variable.set(False)
63        self.root.destroy()
64
65    def end_taken(self):
66        self.integer_variable.set(True)
67        self.root.destroy()
68
69
70class Interruption:
71    def __init__(self, app):
72        if app == False:
73            self.root = tk.Tk()
74            self.label = tk.Label(
75                self.root,
76                text="処理を中断しました\n再度処理を行いたい場合は下記のどちらかを実行してください",
77                font=("Helvetica", 20),
78            ).pack()
79            self.button = tk.Button(
80                self.root, text="Quit", command=self.root.destroy
81            ).pack()
82        else:
83            self.root = tk.Tk()
84            self.root.destroy()
85
86
87if __name__ == "__main__":
88    GUI = GUI()
89    GUI.root.mainloop()
90
91    num = GUI.integer_variable.get()
92    Interruption(num).root.mainloop()

試したこと

＜調査・試したこと＞
・thread関数による別スレッドの作成＆実行
→　問題②
参照したURL：
・https://www.pythontutorial.net/python-concurrency/python-stop-thread/　
→　解決の糸口になりそうなだと思ったURL。しかし、エラーが出て実行不可だった
・https://tsudango-tech.com/88/
・https://teratail.com/questions/76909
・https://magazine.techacademy.jp/magazine/22815
・https://www.pythontutorial.net/python-concurrency/python-stop-thread/
・https://docs.python.org/ja/3/library/threading.html＞
・https://magazine.techacademy.jp/magazine/22221
・https://ja.stackoverflow.com/questions/89069/python%E3%81%AB%E3%81%8A%E3%81%84%E3%81%A6-threading%E3%83%9E%E3%83%AB%E3%83%81%E3%83%AC%E3%83%83%E3%83%89%E3%81%A8tkinter%E3%81%A7%E6%83%B3%E5%AE%9A%E9%80%9A%E3%82%8A%E3%81%AE%E5%8B%95%E4%BD%9C%E3%82%92%E3%81%97%E3%81%AA%E3%81%84

補足情報（FW/ツールのバージョンなど）

＜補足＞
・プログラミング勉強中の身であるため、不勉強な所がありますが今後のためにも何かあればご指摘をいただきたいです。
・試作中のため、変数名などは不適当な部分がありますがご容赦ください
・本問題が解決しなかった場合は、諸事情であまりやりたくなかったことですが、すべての処理を同一クラス内で処理することにします
・本プログラムのほとんどはBingのChatGPTで作成しております。不適切な部分や不具合などは自分のできる範囲で修正し、添付プログラムのようにコーディングしました。

「環境」
・Anaconda3、Windows11、jupyter notebook6.5.4、python 3.10.9

行動規範の内容に同意します

回答2件

ベストアンサー

プログラムとか参照リンクとかあまりちゃんと見ていないのですが、いくつか思うことがあるので、書いてみます。

■ 通常のプログラムは1度に1つのことしかできません。
質問処理は時間のかかる処理を中断させたいということですが、その「時間のかかる処理」を実行しているときに「中断させる」動作をすることはできません。
質問のコードはこうなっていますが、

python
1    GUI = GUI()
2    GUI.root.mainloop()   # (1)
3
4    num = GUI.integer_variable.get()
5    Interruption(num).root.mainloop()    # (2)

これを実行すると、(1)のメインループで、 GUIの処理が実行されますが、GUIの処理が終了するまで、次の行には行きません。 GUIの処理を終了すると、 Interruption()の処理に進みます。
この2つは同時に実行されることはありません。

複数の処理を同時にしたいのであれば、質問にもありますが、マルチスレッドやマルチプロセスの仕組みを使う必要があります。

■ GUIのプログラムはGUI部分を1つにするのが基本
GUIのプログラムは1つのUI(ユーザインターフェース)でユーザとやりとりをすべきです。 1つのというのは Tk() で作ったルートウィンドウです。
質問のプログラムは複数のクラスでTk()が使われていますが、これは、1つのプログラムとしての通常の使いかたではありません。 1つのプログラムでGUIの部分は1つの部分(Class)にまとめるべきです。ダイアログなどのウィンドウを作りたいのであればそのクラスの中でToplevelを使って作るなどします。

■ プロセスは外から止められない
プロセスを外から強制的に中断させることは基本的にできません(すべきでない)。プロセスを止めるには、まず、止められる側のプロセスは、定期的に別プロセスからのメッセージをチェックするなど、中断できる仕組みにします。止める側のプロセスは、止めるときはそのメッセージを出すようにします。ここでのメッセージの送受信をプロセス間通信と呼んで、いろいろな方法があります。「プロセス間通信」で調べてみてください。

質問のプログラムの動作とは違いますが、サンプルでプログラム作ってみました。
起動して、開始ボタンを押すと、裏で別プロセスが起動して、ターミナルに10秒毎にメッセージが出ます。中断ボタンを押すと、そのプロセスは止まります。

python
1import tkinter as tk
2import multiprocessing as mp
3import time
4
5class GUI:
6    def __init__(self):  # 外枠となる設計図を作成
7        self.count = 3
8
9        self.root = tk.Tk()
10        self.root.geometry("300x200")
11        self.integer_variable = tk.BooleanVar()
12        self.label = tk.Label(self.root, text="開始前")
13        self.label.pack()
14        self.start = tk.Button(self.root, text="開始", command=self.start_job)
15        self.start.pack()
16        self.stop = tk.Button(self.root, text="中断", command=self.end_job)
17        self.stop.pack()
18        self.label.after(3, self.check)
19
20        self.longprocess = LongTimeJob()
21        with open('interrupt', mode='w') as f:
22            f.write('run')
23        with open('proc_info', mode='w') as pinfo:
24            pinfo.write('not started')
25
26    def get_proc_status(self):
27        state = 'unknown'
28        with open('proc_info', mode='r') as pinfo:
29            state = pinfo.readline().strip()
30        return state
31
32    def check(self):
33        proc_state = self.get_proc_status()
34        if proc_state == 'end':
35            self.label.configure(text='終了しました')
36            self.longprocess.join()
37        elif proc_state == 'running':
38            self.label.configure(text='実行中')
39            self.label.after(3, self.check)
40        else:
41            self.label.configure(text='状態不明')
42            self.label.after(3, self.check)
43
44    def start_job(self):
45        self.label.after(3, self.check)        
46        self.longprocess.run()
47
48    def end_job(self):
49        with open('interrupt', mode='w') as f:
50            f.write('stop')
51
52class LongTimeJob():
53    def __init__(self):
54        self.job = None
55
56    def report_state(self, state):    
57        with open('proc_info', mode='w') as pinfo:
58            pinfo.write(state)
59
60    def longlongjob(self):
61        while True:
62            print('ジョブ実行中')
63            self.report_state('running')
64            time.sleep(10)
65            if self.check_interrupt():
66                print('中断メッセージ受けました。')
67                break
68        self.report_state('end')
69        print('終了します')
70
71    def check_interrupt(self):
72        with open('interrupt', mode='r') as f:
73            ind = f.readline().strip()
74            if ind == 'stop':
75                return True
76        return False
77
78    def run(self):
79        self.job = mp.Process(target=self.longlongjob)        
80        print('長いジョブ開始')
81        self.job.start()
82        self.report_state('running')
83
84    def join(self):
85        print('ジョインしました')
86        self.job.join()
87
88
89if __name__ == "__main__":
90    GUI = GUI()
91    GUI.root.mainloop()

投稿2023/06/11 07:18

編集2023/06/11 07:19

TakaiY

総合スコア12765

YU-TO

2023/06/12 21:14

回答ありがとうございます！とても分かりやすかったです。各項目を心掛けながら、プログラムを改善していきたいと思います！！

行動規範の内容に同意します

time.sleepでわざわざ実行止めないで、
ここでは単純にループを回し、時間経過、または中断ボタンの押下でループを抜けるようにすればいいかと

投稿2023/06/10 22:33

y_waiwai

総合スコア87774

YU-TO

2023/06/11 00:24

class Timeには本来は全く別のプログラム（Webスクレイピングや機械学習など）を想定しており、今回のclass Timeは試験的な意味で導入したものです。あまりにも機械学習などの処理が長期化し、仕方なく中断するしかないという状況で[タイトルの質問]を行いたいというのが背景になっています。なので、いただいた回答が私の求める内容でなくなってしまいます。説明不足・誤解を与えてしまい、大変申し訳ございません。またご指摘などいただければ幸いです。

行動規範の内容に同意します

あなたの回答