andreasandClaude Sonnet 5
9519169b47
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uc9V1v7iXruubpoLiSE24M
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""System ping helpers shared by ping_monitor and the connectivity probe."""
|
|
|
|
import asyncio
|
|
import re
|
|
import sys
|
|
from typing import Dict, List, Optional
|
|
|
|
_INF = float("inf")
|
|
|
|
|
|
class PingUnavailable(Exception):
|
|
"""The system ping binary could not be executed."""
|
|
|
|
|
|
def ping_command(host: str, count: int, timeout: int, family: Optional[int] = None) -> List[str]:
|
|
"""Build the platform ping command. family is 4, 6 or None (let ping decide)."""
|
|
if sys.platform == "win32":
|
|
cmd = ["ping", "-n", str(count), "-w", str(timeout * 1000)]
|
|
if family:
|
|
cmd.append(f"-{family}")
|
|
return cmd + [host]
|
|
if sys.platform == "darwin" and family:
|
|
if family == 6:
|
|
return ["ping6", "-c", str(count), host]
|
|
return ["ping", "-c", str(count), "-W", str(timeout * 1000), host]
|
|
cmd = ["ping", "-c", str(count), "-W", str(timeout)]
|
|
if family:
|
|
cmd.append(f"-{family}")
|
|
return cmd + [host]
|
|
|
|
|
|
def parse_ping_output(output: str) -> Dict[str, float]:
|
|
"""Parse rtt_min/avg/max/loss from ping output (Linux, macOS, BSD)."""
|
|
loss = 100.0
|
|
loss_match = re.search(r"(\d+(?:\.\d+)?)\s*%\s*packet\s*loss", output)
|
|
if loss_match:
|
|
loss = float(loss_match.group(1))
|
|
|
|
# Linux: "rtt min/avg/max/mdev = x/x/x/x ms"; macOS: "round-trip min/avg/max/stddev = ..."
|
|
rtt_match = re.search(
|
|
r"(?:rtt|round-trip)\s+min/avg/max/\S+\s*=\s*([\d.]+)/([\d.]+)/([\d.]+)",
|
|
output,
|
|
)
|
|
if rtt_match:
|
|
return {
|
|
"rtt_min": float(rtt_match.group(1)),
|
|
"rtt_avg": float(rtt_match.group(2)),
|
|
"rtt_max": float(rtt_match.group(3)),
|
|
"loss": loss,
|
|
}
|
|
return {"rtt_min": _INF, "rtt_avg": _INF, "rtt_max": _INF, "loss": loss}
|
|
|
|
|
|
async def run_ping(host: str, count: int, timeout: int, family: Optional[int] = None) -> Dict[str, float]:
|
|
"""Run ping and return parsed metrics.
|
|
|
|
Raises PingUnavailable if the ping binary is missing and asyncio.TimeoutError
|
|
(after killing the process) if it hangs.
|
|
"""
|
|
cmd = ping_command(host, count, timeout, family)
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
except FileNotFoundError as e:
|
|
raise PingUnavailable(str(e)) from e
|
|
try:
|
|
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout * count + 2)
|
|
except asyncio.TimeoutError:
|
|
proc.kill()
|
|
await proc.wait()
|
|
raise
|
|
return parse_ping_output(stdout.decode(errors="replace"))
|