I am working on a module which receives a continuous flow of numerical inputs. The goal is to detect the first time the array of inputs goes above a certain pre-set threshold value. In other words, I need to run a comparison function until the point when the threshold is reached; then that function needs to be "switched off."
My idea was to approach the problem with decorators as I'm aware they can effectively be used for running a function only once and never again, which is a bit similar to what I am trying to achieve.
In the below case the continuous flow of numerical inputs are: 12, 19, 82, 92, 26, .... In this case the expected output would be:
Rand. val: 12
above_threshold returns False
Rand. val: 19
above_threshold returns False
Rand. val: 82
above_threshold returns True
Threshold has been reached!
Comparison function above_threshold shouldn't be called any more.
Rand. val: 92
Rand. val: 26
...
At the moment however above_threshold gets called in every loop and I haven't succeeded "switching off" the function using the decorator.
import time
import random
random.seed(12771)
threshold = 75
def run_until_first_true_reached(f):
"""
Decorator that runs the function f until it first returns True.
After returning True once, it will stop running the wrapped function again.
"""
def wrapper(*args, **kwargs):
# If f is False
if not f(*args, **kwargs):
return f(*args, **kwargs)
# If f is True
else:
print("Threshold has been reached!")
print("Comparison function above_threshold shouldn't be called any more.")
# tried an empty "return" in this line but didn't solve the issue
return wrapper
@run_until_first_true_reached
def above_threshold(value, threshold):
if value > threshold:
print("above_threshold returns True")
return True
else:
print("above_threshold returns False")
return False
# Modelling the continuous stream of inputs
for _ in range(100):
rand_val = random.randint(1,100)
print("Rand. val: ", rand_val)
above_threshold(rand_val, threshold)
time.sleep(1)
I don't know of any way to have the decorator/wrapper not be called once a condition is reached, but it's pretty simple to turn it into a no-op once a condition is reached, which by your first comment is what you seem to want the code to do.
import time
import random
random.seed(12771)
threshold = 75
def run_until_first_true_reached(f):
"""
Decorator that runs the function f until it first returns True.
After returning True once, it will stop running the wrapped function again.
"""
def wrapper(*args, **kwargs):
if not wrapper.reached:
v = f(*args, **kwargs)
# If f is False
if not v:
return v
# If f is True
else:
print("Threshold has been reached!")
print("Comparison function above_threshold shouldn't be called any more.")
wrapper.reached = True
return None # ? or wahtever we want to return once the threshold is reached
wrapper.reached = False
return wrapper
@run_until_first_true_reached
def above_threshold(value, threshold):
if value > threshold:
print("above_threshold returns True")
return True
else:
print("above_threshold returns False")
return False
# Modelling the continuous stream of inputs
for _ in range(100):
rand_val = random.randint(1,100)
print("Rand. val: ", rand_val)
above_threshold(rand_val, threshold)
time.sleep(1)
Result:
Rand. val: 12
above_threshold returns False
Rand. val: 19
above_threshold returns False
Rand. val: 82
above_threshold returns True
Threshold has been reached!
Comparison function above_threshold shouldn't be called any more.
Rand. val: 92
Rand. val: 26
Rand. val: 18
Rand. val: 55
...
The interesting bit here is that you need somewhere to store the state...the fact that the threshold has been reached. The way I do this in decorators is to attach the state to the wrapper function.
I changed your logic a bit so that the wrapped function isn't called twice on each wrapper invocation. This was producing duplicate lines of output that prevented matching your requested output.