-
Notifications
You must be signed in to change notification settings - Fork 17
/
resampling.py
52 lines (39 loc) · 1.42 KB
/
resampling.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# License: MIT
# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
"""Benchmark resampling."""
from datetime import datetime, timedelta, timezone
from timeit import timeit
from typing import Sequence
from frequenz.sdk.timeseries import Sample
from frequenz.sdk.timeseries._resampling import ResamplerConfig, _ResamplingHelper
def nop( # pylint: disable=unused-argument
samples: Sequence[Sample], resampling_period_s: float
) -> float:
"""Return 0.0."""
return 0.0
def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
"""Benchmark the resampling helper."""
helper = _ResamplingHelper(
ResamplerConfig(
resampling_period_s=1.0,
max_data_age_in_periods=3.0,
resampling_function=nop,
initial_buffer_len=samples * 3,
)
)
now = datetime.now(timezone.utc)
def _do_work() -> None:
nonlocal now
for _n_resample in range(resamples):
for _n_sample in range(samples):
now = now + timedelta(seconds=1 / samples)
helper.add_sample(Sample(now, 0.0))
helper.resample(now)
print(timeit(_do_work, number=5))
def _benchmark() -> None:
for resamples in [10, 100, 1000]:
for samples in [10, 100, 1000]:
print(f"{resamples=} {samples=}")
_benchmark_resampling_helper(resamples, samples)
if __name__ == "__main__":
_benchmark()