简体ä¸ć–‡ | English
A high-performance network ping library built with Rust and exposed to Python.
This package provides fast and reliable ping functionality with both synchronous and asynchronous interfaces. By leveraging Rust's performance and safety guarantees, ping-rs
offers an efficient alternative to traditional Python ping implementations.
uv add ping-rs
Note: If you encounter the error
RuntimeError: Failed to start ping: Could not detect ping.
, please install the ping utility first:# On Debian/Ubuntu sudo apt-get install iputils-ping
from ping_rs import ping_once
# Simple ping (synchronous)
result = ping_once("google.com")
if result.is_success():
print(f"Ping successful! Latency: {result.duration_ms} ms")
else:
print("Ping failed")
import asyncio
from ping_rs import ping_once_async, ping_multiple_async
async def ping_test():
# Single ping asynchronously
result = await ping_once_async("google.com")
if result.is_success():
print(f"Ping successful! Latency: {result.duration_ms} ms")
else:
print("Ping failed")
# Multiple pings asynchronously
results = await ping_multiple_async("google.com", count=5)
for i, result in enumerate(results):
if result.is_success():
print(f"Ping {i+1}: {result.duration_ms} ms")
else:
print(f"Ping {i+1}: Failed")
# Run the async function
asyncio.run(ping_test())
from ping_rs import ping_multiple
# Multiple pings (synchronous)
results = ping_multiple("google.com", count=5)
for i, result in enumerate(results):
if result.is_success():
print(f"Ping {i+1}: {result.duration_ms} ms")
else:
print(f"Ping {i+1}: Failed")
from ping_rs import ping_multiple
# Multiple pings with timeout (will stop after 3 seconds)
results = ping_multiple("google.com", count=10, timeout_ms=3000)
print(f"Received {len(results)} results before timeout")
import time
from ping_rs import create_ping_stream
# Create a non-blocking ping stream
stream = create_ping_stream("google.com")
# Process results as they arrive
while stream.is_active():
result = stream.try_recv()
if result is not None:
if result.is_success():
print(f"Ping: {result.duration_ms} ms")
else:
print("Ping failed")
time.sleep(0.1) # Small delay to avoid busy waiting
from ping_rs import create_ping_stream
# Create a ping stream with a maximum number of 5 pings
stream = create_ping_stream("google.com", count=5)
# Process results using for loop (blocks until each result is available)
for i, result in enumerate(stream):
if result.is_success():
print(f"Ping {i+1}: {result.duration_ms} ms")
else:
print(f"Ping {i+1}: Failed with {result.type_name}")
ping_once(target, timeout_ms=5000, interface=None, ipv4=False, ipv6=False)
: Execute a single ping operation synchronouslyping_once_async(target, timeout_ms=5000, interface=None, ipv4=False, ipv6=False)
: Execute a single ping operation asynchronouslyping_multiple(target, count=4, interval_ms=1000, timeout_ms=None, interface=None, ipv4=False, ipv6=False)
: Execute multiple pings synchronouslyping_multiple_async(target, count=4, interval_ms=1000, timeout_ms=None, interface=None, ipv4=False, ipv6=False)
: Execute multiple pings asynchronouslycreate_ping_stream(target, interval_ms=1000, interface=None, ipv4=False, ipv6=False, count=None)
: Create a non-blocking ping stream
Represents the result of a ping operation.
duration_ms
: Get the ping duration in milliseconds (None if not successful)line
: Get the raw output line from the ping commandexit_code
: Get the exit code if this is a PingExited result, or None otherwisestderr
: Get the stderr output if this is a PingExited result, or None otherwisetype_name
: Get the type name of this PingResult (Pong, Timeout, Unknown, or PingExited)is_success()
: Check if this is a successful ping resultis_timeout()
: Check if this is a timeout resultis_unknown()
: Check if this is an unknown resultis_exited()
: Check if this is a ping process exit resultto_dict()
: Convert this PingResult to a dictionary
High-level ping interface.
__init__(target, interval_ms=1000, interface=None, ipv4=False, ipv6=False)
: Initialize a Pingerping_once()
: Execute a single ping synchronouslyping_stream(count=None)
: Execute multiple pings asynchronously
High-level async ping interface.
__init__(target, interval_ms=1000, interface=None, ipv4=False, ipv6=False)
: Initialize an AsyncPingerping_once()
: Execute a single ping asynchronouslyping_multiple(count=4, timeout_ms=None)
: Execute multiple pings asynchronously
Non-blocking ping stream processor.
try_recv()
: Try to receive the next ping result without blockingrecv()
: Receive the next ping result, blocking if necessaryis_active()
: Check if the stream is still active__iter__
and__next__
: Support for using PingStream as an iterator in a for loop
Async ping stream processor with native async/await support.
__init__(target, interval_ms=1000, interface=None, ipv4=False, ipv6=False, max_count=None)
: Initialize an AsyncPingStream__aiter__()
: Return self as an async iterator__anext__()
: Get the next ping result asynchronously
from ping_rs import ping_once
# Using pattern matching (Python 3.10+)
result = ping_once("google.com")
match result:
case result if result.is_success():
print(f"Success: {result.duration_ms} ms")
case result if result.is_timeout():
print("Timeout")
case result if result.is_unknown():
print(f"Unknown response: {result.line}")
case result if result.is_exited():
print(f"Ping process exited with code {result.exit_code}")
print(f"Error message: {result.stderr}")
case _:
print("Unexpected result type")
# Converting results to dictionaries for data processing
result = ping_once("google.com")
result_dict = result.to_dict()
print(result_dict) # {'type': 'Pong', 'duration_ms': 15.2, 'line': 'Reply from...'}
import asyncio
from ping_rs import AsyncPingStream
async def ping_async_stream():
# Create an async ping stream with a maximum of 5 pings
stream = AsyncPingStream("google.com", interval_ms=1000, max_count=5)
# Process results using async for loop
async for result in stream:
if result.is_success():
print(f"Ping successful: {result.duration_ms} ms")
else:
print(f"Ping failed: {result.type_name}")
# Run the async function
asyncio.run(ping_async_stream())
PingResult can be one of the following types:
-
Pong - Successful ping response
duration_ms
- Ping duration in millisecondsline
- Raw output line from ping command
-
Timeout - Ping timeout
line
- Raw output line with timeout information
-
Unknown - Unrecognized ping response
line
- Raw output line that couldn't be parsed
-
PingExited - Ping process exited unexpectedly
exit_code
- Exit code of the ping processstderr
- Error output from the ping process
The package includes a comprehensive test suite in the tests
directory. To run the tests:
# Run all tests
cd /path/to/ping-rs
python -m tests.run_all_tests
To build the package from source:
cd /path/to/ping-rs
maturin develop
This package uses the following Rust libraries:
- pinger: Provides a cross-platform way to execute ping commands and parse their output. Currently developed as part of the gping project.
- winping: Enables native ICMP ping functionality on Windows platforms without relying on external commands.
MIT License