(name: str, text: str)
| 234 | |
| 235 | |
| 236 | def parse_csv(name: str, text: str) -> dict[str, Any]: |
| 237 | reader = csv.DictReader(text.splitlines()) |
| 238 | series: list[dict[str, Any]] = [] |
| 239 | total_requests = 0.0 |
| 240 | success_weight = 0.0 |
| 241 | |
| 242 | for row in reader: |
| 243 | requests = to_float(row.get("requests")) or 0.0 |
| 244 | success_rate = clamp_01(to_float(row.get("success_ratio"))) |
| 245 | point = { |
| 246 | "minute": row.get("minute"), |
| 247 | "label": row.get("minute"), |
| 248 | "requests": requests, |
| 249 | "tpm": to_float(row.get("tpm")) or requests, |
| 250 | "successRate": success_rate, |
| 251 | "p50Ms": to_float(row.get("p50_ms")), |
| 252 | "p95Ms": to_float(row.get("p95_ms")), |
| 253 | "p99Ms": to_float(row.get("p99_ms")), |
| 254 | "apdex": clamp_01(to_float(row.get("apdex"))), |
| 255 | } |
| 256 | total_requests += requests |
| 257 | if success_rate is not None: |
| 258 | success_weight += success_rate * requests |
| 259 | series.append(point) |
| 260 | |
| 261 | overview = { |
| 262 | "requestsTotal": total_requests or None, |
| 263 | "throughputPerMinute": (total_requests / len(series)) if series else None, |
| 264 | "successRate": (success_weight / total_requests) if total_requests else None, |
| 265 | "p50Ms": last_defined(series, "p50Ms"), |
| 266 | "p95Ms": last_defined(series, "p95Ms"), |
| 267 | "p99Ms": last_defined(series, "p99Ms"), |
| 268 | "peakTpm": max((point.get("tpm") or 0) for point in series) if series else None, |
| 269 | "apdex": mean([point.get("apdex") for point in series]), |
| 270 | } |
| 271 | |
| 272 | return { |
| 273 | "kind": "minute_csv", |
| 274 | "durationMs": len(series) * 60 * 1000 if series else None, |
| 275 | "overview": overview, |
| 276 | "thresholds": [], |
| 277 | "checks": [], |
| 278 | "series": series, |
| 279 | "stats": build_stats(overview, series), |
| 280 | } |
| 281 | |
| 282 | |
| 283 | def parse_ndjson(name: str, text: str) -> dict[str, Any]: |
no test coverage detected