[Python] How to diagnose timedelta in each generator in pipeline?

Objection


I am creating the simple Pipeline class for diagnose callables's process. When I apply to the class for generator functions, it cannot track the intermediate part of timedelta. This is because the generators are only processed during evaluation.

Question


How do you track the execution time for each generator?

I think that the new wrapper Generator class is needed to track each generator execution like a wrapper for the function.

Any ideas?

Prerequisite imports:


from typing import Callable, Generator, Any, Iterable

• Other imports are shown in the sample codes below.
• Python version: 3.10

Codes

1. The Pipeline class

1-1. Construction


Firstly I created a special Iterable to fill in the gaps for zipping unbalanced pairs (for the Pipeline class). This is usually unnecessary, but if there are any errors or interruptions in the middle of the pipeline, the information about the callables may not include some of the Results:

class InfiniteTail(Iterable):
"""
`InfiniteTail` generates `tailed_value` instead of Iterable `obj` if raised StopIteration infinitely.

example usage:

>>> list(zip((1,3,4,5), InfiniteTail((2,3), 0)))
[(1, 2), (3, 3), (4, 0), (5, 0)]
"""
def __init__(self, obj: Iterable, tailed_value: Any = None):
self.__iter = obj.__iter__()
self.tailed_value = tailed_value

def __next__(self):
try:
value = self.__iter.__next__()
except StopIteration:
value = self.tailed_value
return value

def __iter__(self):
return self


I created a dataclass to store a Result for each Callable in the Pipeline class:

from dataclasses import dataclass
@dataclass
class Result:
"""
`Result` for expressing result for each `Callable` in `Pipeline` class.
"""
args: Any = None
kwargs: Any = None
result: Any = None
done: bool = False
ex: Exception = None


I also defined a wrapper for it:

from functools import wraps
def record_result(f: Callable, results: list[Result]) -> Callable:
"""
`record_result` records a `f`'s result and store it in `results` list.
"""
@wraps(f)
def inner(*args, **kwargs):
try:
ret = f(*args, **kwargs)
results.append(Result(
args=args, kwargs=kwargs,
result=ret, done=True, ex=None))
except Exception as ex:
results.append(Result(
args=args, kwargs=kwargs,
result=None, done=False, ex=None))
raise # throw
return ret
return inner


Finally, I created the "Pipeline" class.

from functools import reduce
from IPython.lib.pretty import pretty # for pretty printing with long list/dict etc.
class Pipeline:
"""
`Pipeline` is generalized the flow of `Callable`s.
And some diagnose functionality.

- sample:
>>> def load_data(n):
>>> return list(range(n))
>>> def filtering(xs, upper=50):
>>> return list(x for x in xs if x < upper)
>>> def aggregate(xs):
>>> return sum(xs)
>>> p = Pipeline()
>>> print(p)
load_data->filtering->aggregate
>>> p(100)
1225
>>> print(p)
(100,)->load_data[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]]->filtering[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]]->aggregate[1225]
"""
@classmethod
def pretty_repr(cls, obj: Any) -> str:
return pretty(obj, max_seq_length=10)

def __init__(self, fs: list[Callable]):
self._fs = fs
self._executions = []
"""
`executions` contains results of each execution.
"""
self.__repr_mode = False
"""
False: only show func names, True: show with the last execution info,
"""
def __call__(self, stream: Any) -> Any:
results = []
try:
# reduce self._fs (callables) and get/record result and information.
ret = reduce(lambda s, f: record_result(f, results)(s), self._fs, stream)
except Exception as ex:
raise # throw
finally:
self._executions.append(results)
return ret

def _set_repr_mode(self):
"""
`_set_repr_mode` express show information or not.
"""
if len(self._executions) > 0:
self.__repr_mode = True # show information
else:
self.__repr_mode = False # do not show info, because there is no info.

def _repr_source(self) -> Iterable:
"""
source iterable for `__repr__`
"""
source = self._fs
if self.__repr_mode:
source = list(zip(source, InfiniteTail(self._executions[-1], Result())))
return source
def _element_repr(self, element: Any) -> str:
"""
element repr expression for `__repr__`
"""
if self.__repr_mode:
expr = f"{element[0].__name__}[{self.pretty_repr(element[1].result)}]"
else:
expr = element.__name__
return expr
def __repr__(self) -> str:
"""
`__repr__` returns `__repr_source` converted with `__element_repr`
joined by the delimiter (arrow sign : "->")
"""
self._set_repr_mode() # judge repr mode
exprs = [] # expression for each callable
if self.__repr_mode:
# get the last execution info
results = self._executions[-1]
# get the first result in the info if exists
first : Result = next(iter(results), None)
# if there is a result, show a first argument before the the expressions.
if first is not None:
exprs.append(self.pretty_repr(first.args)) # append the arguments expr
# append the representation of each element to exprs
for element in self._repr_source():
exprs.append(self._element_repr(element))
# return arrow-joined exprs
return "->".join(exprs)

1-2. Test the Pipeline class


Sample code to test the Pipeline class:

def load_data(n: int) -> list:
return list(range(n))
def filtering(xs: list, upper=50) -> list:
return list(x for x in xs if x < upper)
def aggregate(xs: Iterable) -> int:
return sum(xs)

p = Pipeline([load_data, filtering, aggregate])
# test title
print(" --- test for the `Pipeline` class --- ")
# show definition
print(f" definition : {p=}")
# evaluation phase
print(f" evaluation : {p(100)=}")
# show observation from last result
print(f" observation : {p=}")


And the result:


--- test for the `Pipeline` class ---
definition : p=load_data->filtering->aggregate
evaluation : p(100)=1225
observation : p=(100,)->load_data[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]]->filtering[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]]->aggregate[1225]


OK, this is what I expected.

2. The MeasuredPipeline class

2-1. Construction


Next, I made a simple timer to figure out how long each Callables take:

from datetime import datetime, timedelta
class Timer:
"""
simple timer class.

example:
>>> t = Timer()
>>> a = (t.start(), list(range(9_000_000)), t.stop())
>>> t.elapsed()
datetime.timedelta(microseconds=169157)
"""
_zero_delta = timedelta()
def __init__(self):
self._begin = None
self._end = None
self._delta = self._zero_delta
def start(self):
self._begin = datetime.now()
def stop(self):
self._end = datetime.now()
def elapsed(self) -> timedelta:
self._delta = self._end - self._begin \
if self._begin is not None and self._end is not None \
else self._zero_delta
return self._delta


To diagnose callables, I made a wrapper too:

from functools import wraps
def timeit_f(f: Callable, measurements: list[timedelta]) -> Callable:
"""
`timeit_f` diagnose a process time for retrieving a result from f
and append the time to the `measurements` list.
"""
t = Timer()
@wraps(f)
def inner(*args, **kwargs):
t.start()
ret = f(*args, **kwargs)
t.stop()
measurements.append(t.elapsed())
return ret
return inner


I created a special mean function for timedelta:

def timedelta_mean(iterable: Iterable[timedelta], default: timedelta = timedelta()) -> timedelta:
"""
`timedelta_mean` returns the mean of the timedelta in `iterable`.
If there is no element, it returns the `default` value instead.
"""
return sum(iterable, timedelta())/len(iterable) if len(iterable) > 0 else default


Finally, I created the MeasuredPipeline to measure the time it took for each Callable to complete:

class MeasuredPipeline(Pipeline):
"""
`MeasuredPipeline` is generalized the flow of `Callable`s.
And some diagnose functionality. (elapsed time added)

- sample:
>>> def load_data(n):
>>> return list(range(n))
>>> def filtering(xs, upper=1_0000):
>>> return list(x for x in xs if x < upper)
>>> def aggregate(xs):
>>> return sum(xs)
>>> p = Pipeline()
>>> print(p)
load_data->filtering->aggregate
>>> p(1_000_000)
49995000
>>> for _ in range(4):
>>> p(1_000_000)
>>> print(p)
>>> # ([ the mean of `timedelta`s ])@[ the count of executions ]
(1000000,)->load_data[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]](0:00:00.024525)@5->filtering[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]](0:00:00.027953)@5->aggregate[49995000](0:00:00)@5
"""
def __init__(self, fs: list[Callable]):
self._measurements_list = [[] for _ in fs]
fs = [timeit_f(f, l) for f, l in zip(fs, self._measurements_list)]
super().__init__(fs)

def _repr_source(self) -> Iterable:
source = super()._repr_source()
return list(zip(source, self._measurements_list))

def _element_repr(self, element: Any) -> str:
elem, l = element
expr = super()._element_repr(elem)
return f"{expr}({timedelta_mean(l)})@{len(l)}"

2-2. Test the MeasuredPipeline class


Let's take a look at the following. Here's some sample code for the new class. This class is similar to the previous one, but it shows the elapsed average time for each caller.

from functools import partial
# wraps for keeping a callable's `__name__` attribute (partial object do not have that attrib)
mp = MeasuredPipeline([load_data, wraps(filtering)(partial(filtering, upper=1_0000)), aggregate])

print("\n\n")
# test title
print(" --- test for the `MeasuredPipeline` class --- ")
# show definition
print(f" definition : {mp=}")
# test count
test_count = 5
for i, _ in enumerate(range(test_count)):
# evaluation phase
print(f"evaluation@{i+1} : {mp(1_000_000)=}")
# show observation from last result
print(f" observation : {mp=}")


And the result:

--- test for the MeasuredPipeline class ---
definition : mp=load_data(0:00:00)@0->filtering(0:00:00)@0->aggregate(0:00:00)@0
evaluation@1 : mp(1_000_000)=49995000
evaluation@2 : mp(1_000_000)=49995000
evaluation@3 : mp(1_000_000)=49995000
evaluation@4 : mp(1_000_000)=49995000
evaluation@5 : mp(1_000_000)=49995000
observation : mp=(1000000,)->load_data[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]@5->filtering[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, ...]@5
->aggregate49995000@5
Good.
​

2-3. Test the MeasuredPipeline class (with generators)


Then I test the class for generator version callables:

def load_data_gen(n: int) -> Generator:
"""generator version of `load_data`"""
return range(n)
def filtering_gen(xs: Generator, upper=1_0000) -> Generator:
"""generator version of `filtering_gen`"""
return (x for x in xs if x < upper)

mp = MeasuredPipeline([load_data_gen, filtering_gen, aggregate])


print("\n\n")
# test title
print(" --- test for the `MeasuredPipeline` class (generator version) --- ")
# show definition
print(f" definition : {mp=}")
# test count
test_count = 5
for i, _ in enumerate(range(test_count)):
print(f"evaluation@{i+1} : {mp(1_000_000)=}")
# show observation from last result
print(f" observation : {mp=}")


Then the result

--- test for the MeasuredPipeline class (generator version) ---
definition : mp=load_data_gen(0:00:00)@0->filtering_gen(0:00:00)@0->aggregate(0:00:00)@0
evaluation@1 : mp(1_000_000)=49995000
evaluation@2 : mp(1_000_000)=49995000
evaluation@3 : mp(1_000_000)=49995000
evaluation@4 : mp(1_000_000)=49995000
evaluation@5 : mp(1_000_000)=49995000
observation : mp=(1000000,)->load_data_genrange(0, 1000000)@5->filtering_gen<generator object filtering_gen.. at 0x000001FA608DEB20>@5->aggregate49995000@5
​

The first two timedelta means are zeros because the generators are only processed during evaluation. Because of the aggregation phase evaluation, all processing focuses on the last part of the pipeline.

Continue reading...