API Reference
Netspecs can be used as a Python library in addition to the CLI.
Core Modules
netspecs.core.latency
netspecs.core.latency
Latency testing module for netspecs.
LatencyResult
dataclass
Result of a latency test.
status
property
Get status indicator based on results.
to_dict()
Convert result to dictionary.
test_latency(host, count=20, timeout=2)
Test latency to a host using ping.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str
|
Target host to ping |
required |
count
|
int
|
Number of ping packets to send |
20
|
timeout
|
int
|
Timeout in seconds for each ping |
2
|
Returns:
| Type | Description |
|---|---|
LatencyResult
|
LatencyResult with ping statistics |
test_latency_batch(hosts, count=20, timeout=2)
Test latency to multiple hosts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hosts
|
list[str]
|
List of target hosts to ping |
required |
count
|
int
|
Number of ping packets to send per host |
20
|
timeout
|
int
|
Timeout in seconds for each ping |
2
|
Returns:
| Type | Description |
|---|---|
list[LatencyResult]
|
List of LatencyResult objects |
Quick Example
from netspecs.core.latency import test_latency, test_latency_batch
# Single host
result = test_latency("8.8.8.8", count=10)
print(f"Avg latency: {result.avg_ms}ms")
print(f"Packet loss: {result.packet_loss_percent}%")
# Multiple hosts
hosts = ["8.8.8.8", "1.1.1.1", "google.com"]
results = test_latency_batch(hosts, count=20)
for r in results:
print(f"{r.host}: {r.avg_ms}ms, {r.status}")
netspecs.core.speed
netspecs.core.speed
Speed testing module for netspecs.
SpeedResult
dataclass
Result of a speed test.
to_dict()
Convert result to dictionary.
test_bandwidth_variability(iterations=5, test_url='http://speedtest.tele2.net/10MB.zip', size_mb=10)
Test bandwidth variability over multiple iterations.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
iterations
|
int
|
Number of test iterations |
5
|
test_url
|
str
|
URL to use for testing |
'http://speedtest.tele2.net/10MB.zip'
|
size_mb
|
int
|
Size of the test file in MB |
10
|
Returns:
| Type | Description |
|---|---|
dict
|
Dictionary with variability statistics |
test_speed(include_ookla=True, http_tests=None)
Run comprehensive speed tests.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_ookla
|
bool
|
Whether to run Ookla Speedtest CLI if available |
True
|
http_tests
|
Optional[list[dict]]
|
Custom list of HTTP test URLs (uses defaults if None) |
None
|
Returns:
| Type | Description |
|---|---|
SpeedResult
|
SpeedResult with all test results |
Quick Example
from netspecs.core.speed import test_speed, test_bandwidth_variability
# Full speed test
result = test_speed(include_ookla=True)
print(f"HTTP Download: {result.http_download_mbps} Mbps")
if result.ookla_available:
print(f"Ookla Download: {result.ookla_download_mbps} Mbps")
# Variability test
variability = test_bandwidth_variability(iterations=5)
print(f"Speed range: {variability['min_mbps']}-{variability['max_mbps']} Mbps")
netspecs.core.jitter
netspecs.core.jitter
Jitter measurement module for netspecs.
JitterResult
dataclass
Result of a jitter measurement.
status
property
Get status indicator based on jitter.
to_dict()
Convert result to dictionary.
measure_jitter(host, duration=60, interval=1.0)
Measure jitter (latency variation) over time.
Jitter is calculated as the standard deviation of latency samples, which is the proper definition used in networking (not max - min).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str
|
Target host to ping |
required |
duration
|
int
|
Measurement duration in seconds (default increased to 60) |
60
|
interval
|
float
|
Interval between pings in seconds |
1.0
|
Returns:
| Type | Description |
|---|---|
JitterResult
|
JitterResult with jitter statistics including percentiles |
measure_jitter_batch(hosts, duration=30, interval=1)
Measure jitter for multiple hosts sequentially.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hosts
|
list[str]
|
List of target hosts |
required |
duration
|
int
|
Measurement duration per host in seconds |
30
|
interval
|
int
|
Interval between pings in seconds |
1
|
Returns:
| Type | Description |
|---|---|
list[JitterResult]
|
List of JitterResult objects |
Quick Example
from netspecs.core.jitter import measure_jitter
result = measure_jitter("8.8.8.8", duration=30, interval=1)
print(f"Jitter: {result.jitter_ms}ms")
print(f"Range: {result.min_ms}-{result.max_ms}ms")
print(f"Samples: {result.samples}")
netspecs.core.connectivity
netspecs.core.connectivity
Basic connectivity testing module for netspecs.
ConnectivityResult
dataclass
Result of a connectivity check.
to_dict()
Convert result to dictionary.
check_connectivity(host, timeout=2)
Check basic connectivity to a host with a single ping.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str
|
Target host to check |
required |
timeout
|
int
|
Timeout in seconds |
2
|
Returns:
| Type | Description |
|---|---|
ConnectivityResult
|
ConnectivityResult indicating if host is reachable |
check_connectivity_batch(hosts, timeout=2)
Check connectivity to multiple hosts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hosts
|
list[str]
|
List of target hosts to check |
required |
timeout
|
int
|
Timeout in seconds per host |
2
|
Returns:
| Type | Description |
|---|---|
list[ConnectivityResult]
|
List of ConnectivityResult objects |
get_connectivity_summary(results)
Get summary statistics for connectivity results.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
results
|
list[ConnectivityResult]
|
List of ConnectivityResult objects |
required |
Returns:
| Type | Description |
|---|---|
dict
|
Dictionary with summary statistics |
Quick Example
from netspecs.core.connectivity import check_connectivity_batch, get_connectivity_summary
hosts = ["8.8.8.8", "1.1.1.1", "invalid.host"]
results = check_connectivity_batch(hosts)
summary = get_connectivity_summary(results)
print(f"Reachable: {summary['reachable']}/{summary['total_hosts']}")
netspecs.core.prioritization
netspecs.core.prioritization
Prioritization/throttling detection module for netspecs.
PhaseResult
dataclass
Result of a single test phase.
to_dict()
Convert result to dictionary.
PrioritizationResult
dataclass
Result of prioritization detection test.
to_dict()
Convert result to dictionary.
detect_prioritization(host='8.8.8.8', phase_duration=60)
Detect connection prioritization/throttling.
Runs 5 phases: 1. Baseline (idle connection) 2. Light background traffic 3. Medium background traffic 4. Heavy background traffic 5. Return to baseline
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str
|
Target host to test against |
'8.8.8.8'
|
phase_duration
|
int
|
Duration of each phase in seconds |
60
|
Returns:
| Type | Description |
|---|---|
PrioritizationResult
|
PrioritizationResult with analysis |
Quick Example
from netspecs.core.prioritization import detect_prioritization
# This takes ~5 minutes with default settings
result = detect_prioritization(host="8.8.8.8", phase_duration=60)
print(f"Baseline latency: {result.baseline_avg_ms}ms")
print(f"Heavy traffic change: {result.heavy_traffic_change_ms}ms")
if result.throttling_detected:
print(f"Throttling severity: {result.throttling_severity}")
AI Agent
netspecs.agent.analyst
netspecs.agent.analyst
AI diagnostic agent using LiteLLM.
generate_diagnostic_report(results, model=DEFAULT_MODEL, api_key=None, custom_prompt=None)
Generate a narrative diagnostic report from test results using an LLM.
This function uses LiteLLM to support multiple LLM providers (OpenAI, Anthropic, Ollama, etc.) with a unified interface.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
results
|
dict
|
Dictionary of test results from netspecs diagnostics |
required |
model
|
str
|
LLM model to use (default: gpt-5.2) Examples: - "gpt-5.2" (OpenAI) - "gpt-4o" (OpenAI) - "claude-3-sonnet-20240229" (Anthropic) - "ollama/llama2" (Ollama local) |
DEFAULT_MODEL
|
api_key
|
Optional[str]
|
API key for the LLM provider (optional, will use env vars if not provided) |
None
|
custom_prompt
|
Optional[str]
|
Optional custom system prompt to override the default |
None
|
Returns:
| Type | Description |
|---|---|
str
|
Narrative diagnostic report as a string |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no API key is available |
Exception
|
If the LLM call fails |
generate_diagnostic_report_async(results, model=DEFAULT_MODEL, api_key=None)
async
Async version of generate_diagnostic_report.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
results
|
dict
|
Dictionary of test results from netspecs diagnostics |
required |
model
|
str
|
LLM model to use |
DEFAULT_MODEL
|
api_key
|
Optional[str]
|
API key for the LLM provider |
None
|
Returns:
| Type | Description |
|---|---|
str
|
Narrative diagnostic report as a string |
Quick Example
from netspecs.core.latency import test_latency_batch
from netspecs.core.speed import test_speed
from netspecs.agent.analyst import generate_diagnostic_report
# Collect results
latency_results = test_latency_batch(["8.8.8.8", "1.1.1.1"])
speed_result = test_speed()
# Build results dict
results = {
"latency": [r.to_dict() for r in latency_results],
"speed": speed_result.to_dict(),
}
# Generate AI report
report = generate_diagnostic_report(
results,
model="gpt-5.2",
api_key="your-api-key"
)
print(report)
Reporters
netspecs.reporters.console
Rich console output with colors and tables.
from netspecs.reporters.console import ConsoleReporter
reporter = ConsoleReporter()
reporter.report(results)
netspecs.reporters.json
JSON output for machine processing.
from netspecs.reporters.json import JSONReporter
from pathlib import Path
reporter = JSONReporter()
reporter.report(results, output_dir=Path("./results"))
netspecs.reporters.csv
CSV files for spreadsheet analysis.
from netspecs.reporters.csv import CSVReporter
from pathlib import Path
reporter = CSVReporter()
reporter.report(results, output_dir=Path("./results"))
Configuration
netspecs.utils.config
netspecs.utils.config
Configuration management for netspecs.
AIConfig
dataclass
AI report configuration.
BufferbloatTestConfig
dataclass
Bufferbloat test configuration.
Config
dataclass
Main configuration for netspecs diagnostics.
JitterTestConfig
dataclass
Jitter test configuration.
LatencyTestConfig
dataclass
Latency test configuration.
LoggingConfig
dataclass
Logging configuration.
MonitorTestConfig
dataclass
Monitor test configuration.
OutputConfig
dataclass
Output configuration.
PrioritizationTestConfig
dataclass
Prioritization test configuration.
SpeedTestConfig
dataclass
Speed test configuration.
TestsConfig
dataclass
All test configurations.
load_config(config_file=None, env_file=None, api_key=None, model=None, endpoints=None, output_dir=None, output_format=None)
Load configuration from file, environment, and CLI options.
Priority order (highest to lowest): 1. Explicit function arguments 2. Environment variables 3. Config file (netspecs.yaml) 4. Defaults
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config_file
|
Optional[str]
|
Path to config file (auto-detected if not specified) |
None
|
env_file
|
Optional[str]
|
Path to .env file (defaults to .env.local) |
None
|
api_key
|
Optional[str]
|
Explicit API key |
None
|
model
|
Optional[str]
|
LLM model to use |
None
|
endpoints
|
Optional[list[str]]
|
Custom test endpoints |
None
|
output_dir
|
Optional[str]
|
Output directory for logs |
None
|
output_format
|
Optional[str]
|
Output format (console, json, csv) |
None
|
Returns:
| Type | Description |
|---|---|
Config
|
Configured Config instance |
Quick Example
from netspecs.utils.config import load_config, Config
# Load from environment
config = load_config()
# Custom configuration
config = load_config(
api_key="your-key",
model="gpt-4o",
endpoints=["8.8.8.8", "1.1.1.1"],
output_format="json",
)
print(f"Model: {config.model}")
print(f"Endpoints: {config.endpoints}")
Platform Utilities
netspecs.utils.platform
Cross-platform utilities for network operations.
from netspecs.utils.platform import (
get_ping_command,
parse_ping_output,
get_system_info,
)
# Get appropriate ping command for current OS
cmd = get_ping_command("8.8.8.8", count=5)
# ['ping', '-c', '5', '-W', '2', '8.8.8.8'] # on macOS/Linux
# ['ping', '-n', '5', '-w', '2000', '8.8.8.8'] # on Windows
# Get system info
info = get_system_info()
print(f"OS: {info['system']} {info['release']}")