briandconnelly/orbnet

pip install orbnet

uv add orbnet

import asyncio
from orbnet import OrbAPIClient

async def main():
    # Connect to your Orb sensor
    client = OrbAPIClient(host="192.168.0.20", port=7080)
    
    # Get the latest network quality scores
    scores = await client.get_scores_1m()
    
    if scores:
        latest = scores[-1]
        print(f"Orb Score: {latest['orb_score']:.0f}")
        print(f"Responsiveness: {latest['responsiveness_score']:.0f}")
        print(f"Reliability: {latest['reliability_score']:.0f}")
        print(f"Speed: {latest['speed_score']:.0f}")

asyncio.run(main())

from orbnet import OrbAPIClient

client = OrbAPIClient(host="192.168.0.20")

# Get 1-minute scores
scores = await client.get_scores_1m()

# Get responsiveness data (1s, 15s, or 1m granularity)
responsiveness = await client.get_responsiveness(granularity="1s")

# Get web responsiveness (TTFB and DNS)
web_data = await client.get_web_responsiveness()

# Get speed test results
speeds = await client.get_speed_results()

# Get all datasets at once
all_data = await client.get_all_datasets()

async def monitor_network():
    client = OrbAPIClient(host="192.168.0.20")
    
    # Poll for new data every 10 seconds
    async for records in client.poll_dataset(
        dataset_name="responsiveness_1s",
        interval=10.0
    ):
        if records:
            latest = records[-1]
            latency_ms = latest['latency_avg_us'] / 1000
            print(f"Latency: {latency_ms:.1f}ms")

def alert_callback(dataset_name, records):
    for record in records:
        # Alert on high latency
        if record['latency_avg_us'] > 50000:  # 50ms
            print(f"⚠️  High latency: {record['latency_avg_us']}μs")
        
        # Alert on packet loss
        if record['packet_loss_pct'] > 1.0:  # 1%
            print(f"⚠️  Packet loss: {record['packet_loss_pct']:.2f}%")

async def monitor_with_alerts():
    client = OrbAPIClient(host="192.168.0.20")
    
    async for _ in client.poll_dataset(
        dataset_name="responsiveness_1s",
        interval=5.0,
        callback=alert_callback
    ):
        pass  # Callback handles alerts

async def analyze_speeds():
    client = OrbAPIClient(host="192.168.0.20")
    speeds = await client.get_speed_results()
    
    # Convert to Mbps and calculate statistics
    downloads = [s['download_kbps'] / 1000 for s in speeds]
    
    avg_speed = sum(downloads) / len(downloads)
    min_speed = min(downloads)
    max_speed = max(downloads)
    
    print(f"Download Speed Stats:")
    print(f"  Average: {avg_speed:.1f} Mbps")
    print(f"  Minimum: {min_speed:.1f} Mbps")
    print(f"  Maximum: {max_speed:.1f} Mbps")
    
    # Check SLA compliance
    required_mbps = 100
    below_sla = [s for s in downloads if s < required_mbps]
    
    if below_sla:
        pct_below = (len(below_sla) / len(downloads)) * 100
        print(f"⚠️  {pct_below:.1f}% of tests below {required_mbps} Mbps")

async def compare_by_isp():
    client = OrbAPIClient(host="192.168.0.20")
    scores = await client.get_scores_1m()
    
    # Group scores by ISP
    isp_scores = {}
    for record in scores:
        isp = record['isp_name']
        if isp not in isp_scores:
            isp_scores[isp] = []
        isp_scores[isp].append(record['orb_score'])
    
    # Calculate averages
    for isp, scores_list in isp_scores.items():
        avg = sum(scores_list) / len(scores_list)
        print(f"{isp}: {avg:.1f}/100")

client = OrbAPIClient(
    host="192.168.0.20",           # Orb sensor IP/hostname
    port=7080,                      # API port (default: 7080)
    caller_id="my-app",             # Persistent ID for polling
    client_id="MyApp/1.0.0",        # User-Agent identifier
    timeout=30.0,                   # Request timeout in seconds
)

# First request returns all available data
client = OrbAPIClient(host="192.168.0.20", caller_id="monitor-1")
data1 = await client.get_scores_1m()  # Returns: 100 records

# Second request returns only new data
data2 = await client.get_scores_1m()  # Returns: 5 new records

# Development
client = OrbAPIClient(
    host="192.168.0.20",
    client_id="MyApp/dev-0.1.0"
)

# Production with metadata
client = OrbAPIClient(
    host="192.168.0.20",
    client_id="MyApp/1.0.0 (production; host=server-01)"
)

scores = await client.get_scores_1m(format="json")
# Returns: [{"orb_score": 85.5, ...}, {"orb_score": 87.2, ...}]

scores = await client.get_scores_1m(format="jsonl")
# Returns: '{"orb_score": 85.5, ...}\n{"orb_score": 87.2, ...}\n'

# Parse it:
import json
for line in scores.strip().split('\n'):
    record = json.loads(line)
    print(record['orb_score'])

import httpx

try:
    client = OrbAPIClient(host="192.168.0.20", timeout=5.0)
    scores = await client.get_scores_1m()
except httpx.ConnectError:
    print("Unable to connect to Orb sensor")
except httpx.TimeoutException:
    print("Request timed out")
except httpx.HTTPStatusError as e:
    print(f"HTTP error: {e.response.status_code}")

{
  "mcpServers": {
    "orb-net": {
      "command": "uvx",
      "args": [
        "--from",
        "orbnet[mcp]",
        "orbnet-mcp"
      ],
      "env": {
        "ORB_HOST": "<host IP address>",
        "ORB_PORT": "7080"
      }
    }
  }
}