leaky-bucket/poc/py-ratio1024-log2/stream_to_state.py

158 lines
3.8 KiB
Python

import signal
import time
import argparse
from math import log
def main():
args = get_args()
with_(
P=args.p,
D=args.d,
M=args.m,
)
def get_args():
ap = argparse.ArgumentParser()
ap.add_argument("-p", type=str, help="path to write out to", default="/tmp/cbappend.both.txt")
ap.add_argument("-d", type=int, help="milliseconds to retain", default=1000)
ap.add_argument("-m", type=float, help="noise threshold", default=.25)
return ap.parse_args()
def with_(P, D, M):
triggered = CBAppend(P)
released = CBAppend(P)
cb = CBFork(triggered, released)
window = Window(D, M)
previous = {}
while True:
got = readline()
if got:
window.push(got)
report = window.report()
#print(report)
for k in [k for k in report if k in previous]:
if report[k] != previous[k]:
cb.cb(k)(State(True, report[k]))
for k in [k for k in report if not k in previous]:
cb.cb(k)(State(True, report[k]))
for k in [k for k in previous if not k in report]:
cb.cb(k)(State(False, 0))
previous = report
def readline():
def __input(*args):
return input()
def _input(*args):
try:
foo = __input()
except Exception as e:
foo = None
return foo
timeout = 1
signal.signal(signal.SIGALRM, __input)
signal.alarm(timeout)
foo = _input()
return foo
class State():
def __init__(self, active, f):
self.active = active
self.f = f
class Window():
def __init__(self, D, M):
self.D = D
self.M = M
self.w = []
self.n = 1024
def report(self):
self.__pop()
scaled_rates = self.report_scaled_rates()
ttl = sum([scaled_rates[k] for k in scaled_rates])
results = {}
for key in scaled_rates:
results[key] = int(100.0 * scaled_rates[key] / ttl) / 100
return {k:results[k] for k in results if results[k] > self.M}
def report_scaled_rates(self):
cnt = len(self.w)
keys = list(set([i[0] for i in self.w]))
scaled_rates = {}
for key in keys:
count = len([True for i in self.w if i[0] == key])
#scaled_rate = int(log((count / cnt) * 1024.0, 2)+.5)
scaled_rate = int((count / cnt) * 1024.0 + .5)
scaled_rates[key] = scaled_rate
return scaled_rates
def push(self, k):
self.__pop()
self.__push(k)
def pop(self):
self.__pop()
def __push(self, k):
self.w.append((k, self.__now()))
if len(self.w) > self.n:
self.w = self.w[len(self.w)-self.n:]
def __pop(self):
now = self.__now()
self.w = [
i for i in self.w if now - i[1] < self.D
]
def __now(self):
return time.time()*1000
'''
def __cb(self):
new_state = self.q > self.T
if new_state == self.__last_state:
return
self.__last_state = new_state
filledness = int(
100*(
max(
[self.q-self.T, 0]
)/max(
[self.M-self.T, 1]
)
)
)/100.0
if filledness > 1.0:
filledness = 1.0
self.CB(State(new_state, filledness))
'''
class CBAppend():
def __init__(self, path):
self.__path = path
def cb(self, payload):
def cb(state):
with open(self.__path, "a") as f:
f.write(f"{'/' if not state.active else ''}{payload} {state.f}\n")
return cb
class CBFork():
def __init__(self, triggered, released):
self.__triggered = triggered
self.__released = released
def cb(self, payload):
cb_triggered = self.__triggered.cb(payload)
cb_released = self.__released.cb(payload)
def cb(state):
if state.active:
return cb_triggered(state)
return cb_released(state)
return cb
if __name__ == "__main__":
main()