(payload: Any)
| 136 | |
| 137 | |
| 138 | def collect_results(payload: Any) -> dict: |
| 139 | results: dict = defaultdict(lambda: defaultdict(dict)) |
| 140 | benchmarks = payload if isinstance(payload, list) else payload.get("benchmarks", []) |
| 141 | for bench in benchmarks: |
| 142 | benchmark_name = bench.get("benchmark") or bench.get("name", "") |
| 143 | match = BENCHMARK_PATTERN.search(benchmark_name) |
| 144 | if match is None: |
| 145 | continue |
| 146 | metric = bench.get("primaryMetric", {}) |
| 147 | score = float(metric.get("score", bench.get("opsPerSec", 0.0))) |
| 148 | unit = metric.get("scoreUnit", "ops/s") |
| 149 | serializer = match.group("serializer").lower() |
| 150 | if serializer == "fory": |
| 151 | codegen = str(bench.get("params", {}).get("codegen", "unknown")).lower() |
| 152 | serializer = f"fory-codegen={codegen}" |
| 153 | datatype = datatype_key(match.group("datatype")) |
| 154 | operation = match.group("operation").lower() |
| 155 | results[datatype][operation][serializer] = score_to_ops_per_sec(score, unit) |
| 156 | return results |
| 157 | |
| 158 | |
| 159 | def format_tps(value: float, _position) -> str: |
no test coverage detected