Network Latency Monitor (python, written by Claude Code)

Claude Code wrote this python project from the task below. It does not run on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Network Latency Monitor

Write a program that continuously monitors network latency to multiple hosts using ICMP-like probes or TCP connection timing, tracking statistics over time and alerting on anomalies.

FUNCTIONAL REQUIREMENTS:
- Accept one or more hostnames or IP addresses as command-line arguments
- Send periodic probes to each host at a configurable interval via --interval flag (default: 5 seconds)
- Measure round-trip time (RTT) for each probe using TCP connection timing to a specified port (--port flag, default 80) since raw ICMP requires root privileges
- Track running statistics per host: min RTT, max RTT, average RTT, median RTT, standard deviation, jitter (difference between consecutive measurements), and packet loss percentage
- Support configurable monitoring duration via --duration flag (e.g., --duration 60s, --duration 5m, --duration 1h) or run until interrupted with Ctrl+C
- Detect latency anomalies: flag probes where RTT exceeds a configurable threshold via --alert flag (default: 3x the running average RTT)
- Display a live-updating console dashboard showing: host, last RTT, average RTT, min/max, packet loss %, and alert status
- Record all measurements with timestamps for historical analysis
- Save the full monitoring data as a JSON report with --output flag (default: latency_report.json) including per-host statistics and all individual measurements
- Export time-series data as CSV via --export flag for external analysis
- Support reading a list of hosts from a file via --file flag (one host per line with optional label)
- If no hosts are given, monitor a set of well-known public hosts (8.8.8.8, 1.1.1.1, example.com, google.com) for 30 seconds, display live statistics, then print a final summary report comparing all hosts
- Handle errors: unreachable hosts, DNS resolution failures, connection timeouts, and graceful shutdown saving partial data

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

#!/usr/bin/env python3
"""
Network Latency Monitor - Trial 1 (Python)
Monitors network latency to multiple hosts with statistics, anomaly detection,
and a live dashboard using ping3, rich, and numpy.
"""

import time
import signal
import sys
import threading
from collections import deque
from datetime import datetime

import numpy as np
from ping3 import ping
from rich.console import Console
from rich.table import Table
from rich.layout import Layout
from rich.panel import Panel
from rich.live import Live
from rich.text import Text
from rich.bar import Bar


# Default hosts to monitor
DEFAULT_HOSTS = [
    "8.8.8.8",
    "1.1.1.1",
    "208.67.222.222",
    "9.9.9.9",
    "8.8.4.4",
]

# Configuration
PING_INTERVAL = 1.0          # Seconds between pings
HISTORY_SIZE = 100            # Number of samples to keep per host
ANOMALY_ZSCORE_THRESHOLD = 2.5  # Z-score threshold for anomaly detection
TIMEOUT = 2                   # Ping timeout in seconds


class HostStats:
    """Tracks latency statistics for a single host."""

    def __init__(self, host: str, history_size: int = HISTORY_SIZE):
        self.host = host
        self.history_size = history_size
        self.latencies = deque(maxlen=history_size)
        self.total_pings = 0
        self.lost_pings = 0
        self.anomalies = deque(maxlen=20)
        self.last_latency = None
        self.last_ping_time = None
        self.status = "Waiting"

    def record(self, latency_ms: float | None):
        """Record a latency measurement."""
        self.total_pings += 1
        self.last_ping_time = datetime.now()

        if latency_ms is None:
            self.lost_pings += 1
            self.status = "Timeout"
            self.last_latency = None
        else:
            self.latencies.append(latency_ms)
            self.last_latency = latency_ms
            self.status = "OK"
            self._check_anomaly(latency_ms)

    def _check_anomaly(self, latency_ms: float):
        """Detect anomalies using Z-score method."""
        if len(self.latencies) < 10:
            return
        arr = np.array(self.latencies)
        mean = np.mean(arr)
        std = np.std(arr)
        if std == 0:
            return
        zscore = (latency_ms - mean) / std
        if abs(zscore) > ANOMALY_ZSCORE_THRESHOLD:
            self.anomalies.append({
                "time": datetime.now().strftime("%H:%M:%S"),
                "latency": latency_ms,
                "zscore": zscore,
                "mean": mean,
            })

    @property
    def packet_loss(self) -> float:
        if self.total_pings == 0:
            return 0.0
        return (self.lost_pings / self.total_pings) * 100

    @property
    def stats(self) -> dict:
        """Compute statistics from the latency history."""
        if not self.latencies:
            return {"min": 0, "max": 0, "mean": 0, "median": 0, "std": 0, "p95": 0, "p99": 0}
        arr = np.array(self.latencies)
        return {
            "min": float(np.min(arr)),
            "max": float(np.max(arr)),
            "mean": float(np.mean(arr)),
            "median": float(np.median(arr)),
            "std": float(np.std(arr)),
            "p95": float(np.percentile(arr, 95)),
            "p99": float(np.percentile(arr, 99)),
        }


class NetworkLatencyMonitor:
    """Main monitor that pings hosts and manages the dashboard."""

    def __init__(self, hosts: list[str] | None = None):
        self.hosts = hosts or DEFAULT_HOSTS
        self.host_stats: dict[str, HostStats] = {}
        self.console = Console()
        self.running = False
        self.start_time = None
        self.lock = threading.Lock()

        for host in self.hosts:
            self.host_stats[host] = HostStats(host)

    def _ping_host(self, host: str):
        """Ping a single host and record the result."""
        try:
            result = ping(host, timeout=TIMEOUT, unit="ms")
            if result is None or result is False:
                latency_ms = None
            else:
                latency_ms = float(result)
        except Exception:
            latency_ms = None

        with self.lock:
            self.host_stats[host].record(latency_ms)

    def _ping_all_hosts(self):
        """Ping all hosts concurrently."""
        threads = []
        for host in self.hosts:
            t = threading.Thread(target=self._ping_host, args=(host,), daemon=True)
            threads.append(t)
            t.start()
        for t in threads:
            t.join(timeout=TIMEOUT + 1)

    def _build_main_table(self) -> Table:
        """Build the main latency overview table."""
        table = Table(title="Network Latency Monitor", expand=True)
        table.add_column("Host", style="cyan", no_wrap=True)
        table.add_column("Status", justify="center")
        table.add_column("Last (ms)", justify="right")
        table.add_column("Min (ms)", justify="right")
        table.add_column("Avg (ms)", justify="right")
        table.add_column("Max (ms)", justify="right")
        table.add_column("Std (ms)", justify="right")
        table.add_column("P95 (ms)", justify="right")
        table.add_column("Loss %", justify="right")
        table.add_column("Samples", justify="right")

        with self.lock:
            for host in self.hosts:
                hs = self.host_stats[host]
                s = hs.stats

                if hs.status == "OK":
                    status_text = Text("OK", style="green bold")
                elif hs.status == "Timeout":
                    status_text = Text("TIMEOUT", style="red bold")
                else:
                    status_text = Text("WAIT", style="yellow")

                last = f"{hs.last_latency:.1f}" if hs.last_latency is not None else "-"

                loss_style = "green"
                if hs.packet_loss > 5:
                    loss_style = "yellow"
                if hs.packet_loss > 20:
                    loss_style = "red"

                table.add_row(
                    host,
                    status_text,
                    last,
                    f"{s['min']:.1f}" if s['min'] else "-",
                    f"{s['mean']:.1f}" if s['mean'] else "-",
                    f"{s['max']:.1f}" if s['max'] else "-",
                    f"{s['std']:.2f}" if s['std'] else "-",
                    f"{s['p95']:.1f}" if s['p95'] else "-",
                    Text(f"{hs.packet_loss:.1f}%", style=loss_style),
                    str(len(hs.latencies)),
                )

        return table

    def _build_anomaly_panel(self) -> Panel:
        """Build the anomaly detection panel."""
        lines = []
        with self.lock:
            for host in self.hosts:
                hs = self.host_stats[host]
                for a in list(hs.anomalies)[-5:]:
                    lines.append(
                        f"[red][{a['time']}][/red] {host}: "
                        f"{a['latency']:.1f}ms (z={a['zscore']:+.2f}, "
                        f"mean={a['mean']:.1f}ms)"
                    )

        if not lines:
            content = Text("No anomalies detected yet.", style="dim")
        else:
            content = Text.from_markup("\n".join(lines[-10:]))

        return Panel(content, title="Anomaly Detection (Z-score > {})".format(ANOMALY_ZSCORE_THRESHOLD))

    def _build_sparkline(self, host: str) -> str:
        """Build a simple text-based sparkline for a host."""
        blocks = " _.-~*"
        with self.lock:
            hs = self.host_stats[host]
            data = list(hs.latencies)[-40:]

        if len(data) < 2:
            return ""

        mn, mx = min(data), max(data)
        rng = mx - mn if mx != mn else 1
        sparkline = ""
        for v in data:
            idx = int((v - mn) / rng * (len(blocks) - 1))
            sparkline += blocks[idx]
        return sparkline

    def _build_sparkline_panel(self) -> Panel:
        """Build sparkline visualization panel."""
        lines = []
        for host in self.hosts:
            spark = self._build_sparkline(host)
            lines.append(f"[cyan]{host:>18}[/cyan] |{spark}|")

        content = Text.from_markup("\n".join(lines)) if lines else Text("Collecting data...")
        return Panel(content, title="Latency Trends (last 40 samples)")

    def _build_summary_panel(self) -> Panel:
        """Build a summary statistics panel."""
        elapsed = time.time() - self.start_time if self.start_time else 0
        mins, secs = divmod(int(elapsed), 60)
        hrs, mins = divmod(mins, 60)

        total_pings = sum(hs.total_pings for hs in self.host_stats.values())
        total_lost = sum(hs.lost_pings for hs in self.host_stats.values())
        total_anomalies = sum(len(hs.anomalies) for hs in self.host_stats.values())

        summary = (
            f"[bold]Uptime:[/bold] {hrs:02d}:{mins:02d}:{secs:02d}  |  "
            f"[bold]Hosts:[/bold] {len(self.hosts)}  |  "
            f"[bold]Total Pings:[/bold] {total_pings}  |  "
            f"[bold]Lost:[/bold] {total_lost}  |  "
            f"[bold]Anomalies:[/bold] {total_anomalies}"
        )
        return Panel(Text.from_markup(summary), title="Summary")

    def _build_dashboard(self) -> Layout:
        """Compose the full dashboard layout."""
        layout = Layout()
        layout.split_column(
            Layout(name="summary", size=3),
            Layout(name="main", ratio=3),
            Layout(name="bottom", ratio=2),
        )
        layout["bottom"].split_row(
            Layout(name="sparklines"),
            Layout(name="anomalies"),
        )

        layout["summary"].update(self._build_summary_panel())
        layout["main"].update(self._build_main_table())
        layout["sparklines"].update(self._build_sparkline_panel())
        layout["anomalies"].update(self._build_anomaly_panel())

        return layout

    def run(self):
        """Start the monitoring loop with live dashboard."""
        self.running = True
        self.start_time = time.time()

        def handle_signal(sig, frame):
            self.running = False

        signal.signal(signal.SIGINT, handle_signal)

        self.console.print("[bold green]Network Latency Monitor starting...[/bold green]")
        self.console.print(f"Monitoring {len(self.hosts)} hosts. Press Ctrl+C to stop.\n")

        with Live(self._build_dashboard(), console=self.console, refresh_per_second=2) as live:
            while self.running:
                self._ping_all_hosts()
                live.update(self._build_dashboard())
                time.sleep(PING_INTERVAL)

        self.console.print("\n[bold yellow]Monitor stopped.[/bold yellow]")
        self._print_final_report()

    def _print_final_report(self):
        """Print a final summary report."""
        self.console.print("\n[bold]Final Report[/bold]")
        table = Table(title="Latency Summary")
        table.add_column("Host", style="cyan")
        table.add_column("Pings", justify="right")
        table.add_column("Loss %", justify="right")
        table.add_column("Min", justify="right")
        table.add_column("Mean", justify="right")
        table.add_column("Median", justify="right")
        table.add_column("P95", justify="right")
        table.add_column("P99", justify="right")
        table.add_column("Max", justify="right")
        table.add_column("StdDev", justify="right")

        for host in self.hosts:
            hs = self.host_stats[host]
            s = hs.stats
            table.add_row(
                host,
                str(hs.total_pings),
                f"{hs.packet_loss:.1f}%",
                f"{s['min']:.1f}",
                f"{s['mean']:.1f}",
                f"{s['median']:.1f}",
                f"{s['p95']:.1f}",
                f"{s['p99']:.1f}",
                f"{s['max']:.1f}",
                f"{s['std']:.2f}",
            )

        self.console.print(table)


def main():
    import argparse

    parser = argparse.ArgumentParser(description="Network Latency Monitor")
    parser.add_argument(
        "hosts", nargs="*", default=DEFAULT_HOSTS,
        help="Hosts to monitor (IP addresses or hostnames)"
    )
    parser.add_argument(
        "-i", "--interval", type=float, default=PING_INTERVAL,
        help="Ping interval in seconds (default: 1.0)"
    )
    parser.add_argument(
        "-n", "--history-size", type=int, default=HISTORY_SIZE,
        help="Number of latency samples to keep (default: 100)"
    )
    parser.add_argument(
        "-t", "--timeout", type=int, default=TIMEOUT,
        help="Ping timeout in seconds (default: 2)"
    )
    parser.add_argument(
        "-z", "--zscore-threshold", type=float, default=ANOMALY_ZSCORE_THRESHOLD,
        help="Z-score threshold for anomaly detection (default: 2.5)"
    )

    args = parser.parse_args()

    global PING_INTERVAL, HISTORY_SIZE, TIMEOUT, ANOMALY_ZSCORE_THRESHOLD
    PING_INTERVAL = args.interval
    HISTORY_SIZE = args.history_size
    TIMEOUT = args.timeout
    ANOMALY_ZSCORE_THRESHOLD = args.zscore_threshold

    monitor = NetworkLatencyMonitor(hosts=args.hosts)
    monitor.run()


if __name__ == "__main__":
    main()

# Network Latency Monitor (Python - Trial 1)

A network latency monitoring tool that tracks latency to multiple hosts with real-time statistics, anomaly detection using Z-scores, and a live terminal dashboard.

## Dependencies

- **ping3** (4.0.4) - Pure Python ICMP ping implementation
- **rich** (13.7.0) - Rich text and beautiful formatting for the terminal dashboard
- **numpy** (1.26.4) - Numerical computations for statistics and anomaly detection

## Setup

```bash
pip install -r requirements.txt
```

## Usage

```bash
# Monitor default hosts (8.8.8.8, 1.1.1.1, etc.)
sudo python monitor.py

# Monitor specific hosts
sudo python monitor.py 8.8.8.8 1.1.1.1 google.com

# Custom interval and timeout
sudo python monitor.py -i 2.0 -t 5 8.8.8.8

# Adjust anomaly detection sensitivity
sudo python monitor.py -z 3.0
```

Note: ICMP ping typically requires root/administrator privileges.

## Features

- Real-time latency monitoring of multiple hosts
- Live terminal dashboard with Rich
- Statistical analysis: min, max, mean, median, P95, P99, standard deviation
- Z-score based anomaly detection
- Packet loss tracking
- Text-based sparkline visualizations
- Final summary report on exit

ping3==4.0.4
rich==13.7.0
numpy==1.26.4