public
Last active

  • Download Gist
afd.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
import math
import time
 
class AccrualFailureDetector(object):
 
SAMPLE_SIZE = 1000
 
def __init__(self):
self.last_time = None
self.intervals = []
 
def add(self, arrival=None):
if not arrival:
raise AttributeError("You must supply an arrival time")
else:
arrival = int(arrival)
 
if not self.last_time:
# Set arbitrary interval, allow suspicion to be lowered by peer.
interval = 500
else:
interval = arrival - self.last_time
 
self.last_time = arrival
self.intervals.append(interval)
if len(self.intervals) > self.SAMPLE_SIZE:
self.intervals.pop(0)
 
def clear(self):
if self.intervals:
del self.intervals[:]
 
def phi(self, current_time=None):
if not current_time:
current_time = int(time.time())
 
# Calculate probability
current_interval = current_time - self.last_time
exp = -1.0 * current_interval / self._interval_mean()
prob = 1 - (1.0 - math.pow(math.e, exp))
 
return -1 * math.log10(prob)
 
def _interval_mean(self):
return sum(self.intervals) / len(self.intervals)

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.