Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, source=None):
Op.__init__(self, source)
self._count = 0
self._sum = 0
def __init__(self, count, source=None):
Op.__init__(self, source)
self._count = count
self._q = deque()
def __init__(self, func, initializer=NO_VALUE, source=None):
Op.__init__(self, source)
self._func = func
self._initializer = initializer
self._prev = NO_VALUE
def __init__(self, timeout, source=None):
Op.__init__(self, source)
if source.done():
return
self._timeout = timeout
self._last_time = loop.time()
self._handle = None
self._schedule()
def __init__(self, source=None):
Op.__init__(self, source)
self._values = []
def __init__(self, interval, on_first=False, source=None):
Op.__init__(self, source)
self._interval = interval
self._on_first = on_first
self._last_time = -float('inf')
self._handle = None
def __init__(self, delay, source=None):
Op.__init__(self, source)
self._delay = delay
def __init__(self, timer, source=None):
Op.__init__(self, source)
self._timer = timer
timer.connect(
self._on_timer,
self.on_source_error,
self.on_source_done)
def __init__(self, maximum, interval, cost_func=None, source=None):
Op.__init__(self, source)
self.status_event = Event('throttle_status')
"""
Sub event that emits ``True`` when throttling starts and ``False``
when throttling ends.
"""
self._maximum = maximum
self._interval = interval
self._cost_func = cost_func
self._q = deque() # deque of (args, cost) tuples
self._time_q = deque() # deque of previous emit times
self._cost_q = deque() # deque of costs of previous emits
self._is_throttling = False
def __init__(self, n=None, weight=None, source=None):
Op.__init__(self, source)
self._f1 = weight or 2.0 / (n + 1)
self._f2 = 1 - self._f1
self._prev = NO_VALUE