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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
from __future__ import annotations
from copy import deepcopy
from datetime import datetime
from typing import TYPE_CHECKING, Any
from asv.benchmarks import Benchmarks
from asv.config import Config
from asv.results import Results, iter_results_for_machine
from asv_runner.statistics import get_err
import numpy as np
from contourpy import FillType, LineType
if TYPE_CHECKING:
from asv.benchmark import Benchmark
class Loader:
_config: Config
_benchmarks: Benchmarks
_machine: str
_results: Results
def __init__(self, machine: str | None = None) -> None:
self._config = Config.load()
self._benchmarks = Benchmarks.load(self._config)
if machine is None:
import platform
machine = platform.uname()[1]
latest_results = None
for results in iter_results_for_machine(self._config.results_dir, machine):
if latest_results is None or results.date > latest_results.date:
latest_results = results
if latest_results is None:
raise RuntimeError(f"No results found for machine {machine}")
print(latest_results.commit_hash, datetime.fromtimestamp(latest_results.date/1000.0))
self._results = latest_results
self._machine = machine
def _find_benchmark_by_name(self, name: str) -> Benchmark:
for k, v in self._benchmarks.items():
if k.endswith(name):
return v
raise RuntimeError(f"Cannot find benchmark with name {name}")
@property
def commit(self) -> str:
return self._results.commit_hash # type: ignore[no-any-return]
def get(self, benchmark_name: str, **kwargs: Any) -> dict[str, Any]:
benchmark = self._find_benchmark_by_name(benchmark_name)
param_names = benchmark["param_names"]
params = deepcopy(benchmark["params"])
for name, value in kwargs.items():
index = param_names.index(name)
if isinstance(value, list):
params[index] = [repr(item) for item in value]
else:
params[index] = [repr(value)]
stats = self._results.get_result_stats(benchmark["name"], params)
values = self._results.get_result_value(benchmark["name"], params)
ret = {}
for name, param in zip(param_names, params):
for i, item in enumerate(param):
if isinstance(item, str):
if item[0] == "'" and item[-1] == "'":
item = item[1:-1]
if item.startswith("<FillType"):
item = FillType(int(item[item.index(" "):-1]))
elif item.startswith("<LineType"):
item = LineType(int(item[item.index(" "):-1]))
elif item == "False":
item = False
elif item == "True":
item = True
else:
try:
item = int(item)
except ValueError:
pass
param[i] = item
ret[name] = param[0] if len(param) == 1 else param
if values[0] is None or np.isnan(values[0]):
ret["mean"] = ret["error"] = None
else:
ret["mean"] = values
ret["error"] = [get_err(v, s) for v, s in zip(values, stats)]
return ret
@property
def machine(self) -> str:
return self._machine
|
