140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
from typing import Callable
|
|
|
|
class EzSequence:
|
|
def __init__(self, start_value):
|
|
self.start_value = start_value
|
|
self.value = start_value
|
|
self.index = 0
|
|
self.when_funcs = []
|
|
self.not_when_funcs = []
|
|
self.until_funcs = []
|
|
self.as_long_as_funcs = []
|
|
self.is_last = False
|
|
self.by(1)
|
|
self.consecutive_attempts = 0
|
|
self.safely(100)
|
|
self.is_first_element = True
|
|
self.translate_as(lambda index, value: value)
|
|
def reset(self):
|
|
self.value = self.start_value
|
|
self.index = 0
|
|
self.is_first_element = True
|
|
self.is_last = False
|
|
return self
|
|
def __iter__(self):
|
|
return self
|
|
def __next__(self) -> int:
|
|
if self.is_first_element:
|
|
self.is_first_element = False
|
|
|
|
if self.until_funcs != None and len(self.until_funcs) > 0:
|
|
if any(f(self.index, self.value) for f in self.until_funcs):
|
|
self.is_last = True
|
|
|
|
if ((self.when_funcs != None and len(self.when_funcs) > 0) or (self.not_when_funcs != None and len(self.not_when_funcs) > 0)):
|
|
if (all(f(self.index, self.value) for f in self.when_funcs)) and (not any(f(self.index, self.value) for f in self.not_when_funcs)):
|
|
return self.translate_as_func(self.index, self.value)
|
|
else:
|
|
pass
|
|
else:
|
|
return self.translate_as_func(self.index, self.value)
|
|
|
|
self.index -= 1 # Because it will be incremented below and we are still in the first element.
|
|
|
|
#current_index = self.index
|
|
current_value = self.value
|
|
|
|
# Handle UNTIL
|
|
if self.is_last:
|
|
raise StopIteration
|
|
|
|
# Find next value.
|
|
|
|
self.consecutive_attempts = 1
|
|
next_value = self.increment_func(self.index + 1, current_value)
|
|
next_index = self.index + 1
|
|
|
|
# Keep looking for a new value until we get a value that can be output or until we violate the safety limit.
|
|
if ((self.when_funcs != None and len(self.when_funcs) > 0) or (self.not_when_funcs != None and len(self.not_when_funcs) > 0)):
|
|
while True:
|
|
if (not all(f(next_index, next_value) for f in self.when_funcs)) or (any(f(next_index, next_value) for f in self.not_when_funcs)):
|
|
if self.is_last:
|
|
raise StopIteration
|
|
|
|
# This value will not be output. Find a new one that will.
|
|
self.consecutive_attempts += 1
|
|
if self.use_safety and (self.consecutive_attempts > self.max_consective_attempts):
|
|
raise Exception(f"Safe iteration limit exceeded. {self.max_consective_attempts} attempts exceeded.")
|
|
|
|
next_value = self.increment_func(next_index, next_value)
|
|
|
|
# Stop if we encountered the end of the sequence.
|
|
if self.as_long_as_funcs != None and len(self.as_long_as_funcs) > 0:
|
|
if not all(f(next_index, next_value) for f in self.as_long_as_funcs):
|
|
raise StopIteration
|
|
|
|
if self.until_funcs != None and len(self.until_funcs) > 0:
|
|
if any(f(next_index, next_value) for f in self.until_funcs):
|
|
self.is_last = True
|
|
else:
|
|
break
|
|
|
|
# Stop if we encountered the end of the sequence.
|
|
if self.as_long_as_funcs != None and len(self.as_long_as_funcs) > 0:
|
|
if not all(f(next_index, next_value) for f in self.as_long_as_funcs):
|
|
raise StopIteration
|
|
|
|
# Return the value THIS time, but prepare to stop NEXT time if UNTIL is encountered here.
|
|
if self.until_funcs != None and len(self.until_funcs) > 0:
|
|
if any(f(next_index, next_value) for f in self.until_funcs):
|
|
self.is_last = True
|
|
|
|
self.index + next_index
|
|
self.value = next_value
|
|
|
|
return self.translate_as_func(self.index, self.value)
|
|
|
|
def by(self, increment: Callable[[int, int], int]):
|
|
self.with_increment(lambda index, value: value + increment)
|
|
return self
|
|
|
|
def with_increment(self, increment_func: Callable[[int, int], int] ):
|
|
self.increment_func = increment_func
|
|
return self
|
|
|
|
def until(self, until_func: Callable[[int, int], bool]):
|
|
self.until_funcs.append(until_func)
|
|
return self
|
|
|
|
def as_long_as(self, as_long_as_func: Callable[[int, int], bool]):
|
|
self.as_long_as_funcs.append(as_long_as_func)
|
|
return self
|
|
|
|
def to(self, last_value):
|
|
self.until(lambda index, value: value == last_value)
|
|
return self
|
|
|
|
def when(self, when_func: Callable[[int, int], bool]):
|
|
self.when_funcs.append(when_func)
|
|
return self
|
|
|
|
def not_when(self, not_when_func: Callable[[int, int], bool]):
|
|
self.not_when_funcs.append(not_when_func)
|
|
return self
|
|
|
|
def translate_as(self, translate_as_func: Callable[[int, int], int]):
|
|
self.translate_as_func = translate_as_func
|
|
return self
|
|
|
|
def safely(self, max_attempts: int = 10000):
|
|
self.use_safety = True
|
|
self.max_consective_attempts = max_attempts
|
|
return self
|
|
|
|
def unsafely(self):
|
|
self.use_safety = False
|
|
return self
|
|
|
|
def sequence_from(start_value: int):
|
|
return EzSequence(start_value)
|