diff --git a/src/netsleuth/speed.py b/src/netsleuth/speed.py index 505e1f0..f0f885d 100644 --- a/src/netsleuth/speed.py +++ b/src/netsleuth/speed.py @@ -39,7 +39,10 @@ def parse_server_timing_cfl4(header: str) -> CfL4Stats | None: match = _CFL4_RE.search(header or "") if not match: return None - fields = {k: v[0] for k, v in parse_qs(match.group("query"), keep_blank_values=True).items()} + fields = { + k: v[0] + for k, v in parse_qs(match.group("query"), keep_blank_values=True).items() + } return CfL4Stats( rtt_ms=_us_to_ms(fields.get("rtt")), min_rtt_ms=_us_to_ms(fields.get("min_rtt")), @@ -51,7 +54,9 @@ def parse_server_timing_cfl4(header: str) -> CfL4Stats | None: ) -def bufferbloat_delta(idle_rtt_ms: float | None, loaded_rtts_ms: list[float]) -> float | None: +def bufferbloat_delta( + idle_rtt_ms: float | None, loaded_rtts_ms: list[float] +) -> float | None: if idle_rtt_ms is None or not loaded_rtts_ms: return None loaded = percentile(sorted(loaded_rtts_ms), 95.0) @@ -93,10 +98,17 @@ async def run_speed_cascade( duration_ms = int((time.perf_counter() - began) * 1000) if not result.download_mbps: attempts.append( - TierAttempt(tier=name, ok=False, reason="no throughput measured", duration_ms=duration_ms) + TierAttempt( + tier=name, + ok=False, + reason="no throughput measured", + duration_ms=duration_ms, + ) ) continue - attempts.append(TierAttempt(tier=name, ok=True, reason=None, duration_ms=duration_ms)) + attempts.append( + TierAttempt(tier=name, ok=True, reason=None, duration_ms=duration_ms) + ) result.tier_attempts = attempts return result return SpeedResult(method="none", tier_attempts=attempts) @@ -138,15 +150,21 @@ async def tier_ookla( server_info = payload.get("server") or {} return SpeedResult( method="ookla_bin", - download_mbps=round(mbps(download.get("bytes", 0), (download.get("elapsed", 0) or 0) / 1000), 3), - upload_mbps=round(mbps(upload.get("bytes", 0), (upload.get("elapsed", 0) or 0) / 1000), 3), + download_mbps=round( + mbps(download.get("bytes", 0), (download.get("elapsed", 0) or 0) / 1000), 3 + ), + upload_mbps=round( + mbps(upload.get("bytes", 0), (upload.get("elapsed", 0) or 0) / 1000), 3 + ), server=f"{server_info.get('name', '')} ({server_info.get('location', '')})".strip(), idle_rtt_ms=(payload.get("ping") or {}).get("latency"), server_country=server_info.get("country") or None, ) -async def tier_cloudflare(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: +async def tier_cloudflare( + client: httpx.AsyncClient, cfg: Speedtest, timeout: float +) -> SpeedResult: down_samples: list[tuple[int, float]] = [] cfl4: CfL4Stats | None = None for size in cfg.download_sizes_bytes: @@ -156,12 +174,16 @@ async def tier_cloudflare(client: httpx.AsyncClient, cfg: Speedtest, timeout: fl ) response.raise_for_status() down_samples.append((len(response.content), time.perf_counter() - began)) - cfl4 = parse_server_timing_cfl4(response.headers.get("server-timing", "")) or cfl4 + cfl4 = ( + parse_server_timing_cfl4(response.headers.get("server-timing", "")) or cfl4 + ) up_samples: list[tuple[int, float]] = [] for size in cfg.upload_sizes_bytes: payload = b"\x00" * size began = time.perf_counter() - response = await client.post(f"{cfg.cloudflare_base_url}/__up", content=payload, timeout=timeout) + response = await client.post( + f"{cfg.cloudflare_base_url}/__up", content=payload, timeout=timeout + ) response.raise_for_status() up_samples.append((size, time.perf_counter() - began)) return SpeedResult( @@ -173,7 +195,9 @@ async def tier_cloudflare(client: httpx.AsyncClient, cfg: Speedtest, timeout: fl ) -async def tier_fastcom(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: +async def tier_fastcom( + client: httpx.AsyncClient, cfg: Speedtest, timeout: float +) -> SpeedResult: response = await client.get( cfg.fastcom_api_url, params={"https": "true", "token": cfg.fastcom_api_token, "urlCount": "3"}, @@ -181,30 +205,44 @@ async def tier_fastcom(client: httpx.AsyncClient, cfg: Speedtest, timeout: float ) response.raise_for_status() targets = response.json().get("targets") or [] - samples: list[tuple[int, float]] = [] on_net: bool | None = None + + async def fetch(url: str) -> tuple[int, float]: + began = time.perf_counter() + body = await client.get(url, timeout=timeout) + body.raise_for_status() + return len(body.content), time.perf_counter() - began + + tasks = [] for target in targets: url = target.get("url") if not url: continue location = (target.get("location") or {}).get("country") on_net = on_net or bool(location) - began = time.perf_counter() - body = await client.get(url, timeout=timeout) - body.raise_for_status() - samples.append((len(body.content), time.perf_counter() - began)) + tasks.append(asyncio.create_task(fetch(url))) + + samples = await asyncio.gather(*tasks) if tasks else [] + return SpeedResult( method="fastcom", download_mbps=throughput_from_samples(samples), upload_mbps=None, - server=targets[0].get("url").split("/")[2] if targets and targets[0].get("url") else None, + server=targets[0].get("url").split("/")[2] + if targets and targets[0].get("url") + else None, netflix_oca_onnet=on_net, - server_country=(targets[0].get("location") or {}).get("country") if targets else None, + server_country=(targets[0].get("location") or {}).get("country") + if targets + else None, ) async def tier_ndt7( - client: httpx.AsyncClient, cfg: Speedtest, timeout: float, source_ip: str | None = None + client: httpx.AsyncClient, + cfg: Speedtest, + timeout: float, + source_ip: str | None = None, ) -> SpeedResult: import websockets @@ -225,7 +263,11 @@ async def tier_ndt7( message = await asyncio.wait_for(socket.recv(), timeout=timeout) except (TimeoutError, Exception): break - total += len(message) if isinstance(message, (bytes, bytearray)) else len(message.encode()) + total += ( + len(message) + if isinstance(message, (bytes, bytearray)) + else len(message.encode()) + ) elapsed = time.perf_counter() - began return SpeedResult( method="ndt7", @@ -276,7 +318,11 @@ async def measure_with_bufferbloat( result.bufferbloat_down_ms = bufferbloat_delta(idle_rtt_ms, down_samples) result.bufferbloat_up_ms = bufferbloat_delta(idle_rtt_ms, up_samples) worst_delta = max( - [d for d in (result.bufferbloat_down_ms, result.bufferbloat_up_ms) if d is not None], + [ + d + for d in (result.bufferbloat_down_ms, result.bufferbloat_up_ms) + if d is not None + ], default=None, ) result.bufferbloat_grade = grade_bufferbloat(worst_delta, bands)