I'm trying to create a class that pings an ip address and keeps a record for connected/ not connected times.
Since this class is a part of a GUI, I wish to stop this thread when asked by user.
Found some Q&As regrading this issue, but neither one actually causes thread to stop.
I'm trying to make a method, a part of this class that will stop self.run()
Here's my Pinger
class:
class Pinger(threading.Thread):
def __init__(self, address='', rate=1):
threading.Thread.__init__(self)
self.address = address
self.ping_rate = rate
self.ping_vector, self.last_ping = [], -1
self.start_time, self.last_status = datetime.datetime.now(), []
self.timestamp, self.time_vector = 0, [datetime.timedelta(0)] * 4
def run(self):
self.start_ping()
def start_ping(self):
self.timestamp = datetime.datetime.now()
while True:
ping_result = os.system('ping %s -n 1 >Null' % self.address)
self.ping_vector.append(ping_result)
if self.last_ping != ping_result:
text = ['Reachable', 'Lost']
print(str(self.timestamp)[:-4], self.address, text[ping_result])
round_time_qouta = datetime.datetime.now() - self.timestamp
self.timestamp = datetime.datetime.now()
self.update_time_counter(ping_result, round_time_qouta)
self.last_ping = ping_result
time.sleep(self.ping_rate)
def update_time_counter(self, ping_result=0, time_quota=datetime.timedelta(0)):
"""self.time_vector = [[cons.succ ping time],[cons.not_succ ping time],
[max accum succ ping time],[max accum not_succ ping time] """
p_vec = [0, 1]
self.time_vector[p_vec[ping_result]] += time_quota
if self.time_vector[p_vec[ping_result]].total_seconds() > self.time_vector[
p_vec[ping_result] + 2].total_seconds():
self.time_vector[p_vec[ping_result] + 2] = self.time_vector[p_vec[ping_result]]
self.time_vector[p_vec[ping_result - 1]] = datetime.timedelta(0)
self.last_status = [ping_result, self.chop_milisecond(self.time_vector[ping_result]),
self.chop_milisecond(self.time_vector[ping_result + 2]),
self.chop_milisecond(datetime.datetime.now() - self.start_time)]
print(str(self.timestamp)[:-4], "State: " + ['Received', 'Lost'][ping_result],
" Duration: " + self.last_status[1], " Max Duration: " + self.last_status[2],
"Total time: " + self.last_status[3])
def chop_milisecond(self, time):
return str(time).split('.')[0]
Best Answer
As I was saying in my comment, the easiest way is to use
threading.Event
to signal your thread when it should exit. That way you can expose the event and let other threads set it while you can check for its state from within your thread and exit on request.In your case, it could be as simple as:
Then whenever you want the thread stopped (like from your UI), just call on it:
pinger_instance.kill.set()
and you're done. Keep in mind, tho, that it will take some time for it to get killed due to the blockingos.system()
call and due to thetime.sleep()
you have at the end of yourPinger.start_ping()
method.