Network Latency Monitor (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran 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

[]

# Network Latency Monitor (Python)

Continuously monitors latency to multiple hosts via TCP connection timing and reports RTT statistics, jitter, packet loss, and anomalies.

## Requirements
- Ubuntu 22.04
- Python 3.10+

## Dependencies
- No external dependencies (standard library only)

## Setup
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run
```bash
python3 src/main.py google.com 1.1.1.1 --interval 5 --port 80 --duration 60s
```

With host file and CSV export:
```bash
python3 src/main.py --file hosts.txt --interval 2 --duration 5m --output latency_report.json --export latency_series.csv
```

Alert threshold:
```bash
python3 src/main.py google.com --alert 3x
python3 src/main.py google.com --alert 200
```

## Host file format
```text
8.8.8.8 Google-DNS
1.1.1.1,Cloudflare
example.com
```

## Output
- Live dashboard
- JSON report (`--output`, default `latency_report.json`)
- Optional CSV time-series (`--export`)

# No external dependencies required.
# Uses Python 3.10+ standard library only.

#!/usr/bin/env python3
"""Network latency monitor using TCP connection timing."""

from __future__ import annotations

import argparse
import csv
import json
import math
import signal
import socket
import statistics
import sys
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional


def parse_duration(value: Optional[str]) -> Optional[float]:
    if value is None:
        return None
    value = value.strip().lower()
    if value.endswith("ms"):
        return float(value[:-2]) / 1000.0
    if value.endswith("s"):
        return float(value[:-1])
    if value.endswith("m"):
        return float(value[:-1]) * 60.0
    if value.endswith("h"):
        return float(value[:-1]) * 3600.0
    raise ValueError(f"Invalid duration: {value}")


def parse_alert(value: str) -> tuple[str, float]:
    raw = value.strip().lower()
    if raw.endswith("x"):
        return ("multiplier", float(raw[:-1]))
    return ("absolute", float(raw))


def parse_hosts_file(path: Path) -> list[tuple[str, str]]:
    out: list[tuple[str, str]] = []
    for line_raw in path.read_text(encoding="utf-8").splitlines():
        line = line_raw.strip()
        if not line or line.startswith("#"):
            continue
        if "," in line:
            host, label = line.split(",", 1)
            host = host.strip()
            label = label.strip() or host
            out.append((host, label))
            continue
        parts = line.split()
        host = parts[0]
        label = " ".join(parts[1:]) if len(parts) > 1 else host
        out.append((host, label))
    return out


@dataclass
class Measurement:
    timestamp: str
    success: bool
    rtt_ms: Optional[float]
    error: Optional[str]
    alert: bool


@dataclass
class HostState:
    host: str
    label: str
    sent: int = 0
    lost: int = 0
    last_rtt: Optional[float] = None
    last_alert: bool = False
    rtts: list[float] = field(default_factory=list)
    jitters: list[float] = field(default_factory=list)
    measurements: list[Measurement] = field(default_factory=list)

    def stats(self) -> dict:
        count = len(self.rtts)
        avg = (sum(self.rtts) / count) if count else None
        med = statistics.median(self.rtts) if count else None
        std = statistics.pstdev(self.rtts) if count > 1 else (0.0 if count == 1 else None)
        jitter = (sum(self.jitters) / len(self.jitters)) if self.jitters else None
        return {
            "sent": self.sent,
            "lost": self.lost,
            "success_count": count,
            "min_rtt_ms": min(self.rtts) if count else None,
            "max_rtt_ms": max(self.rtts) if count else None,
            "avg_rtt_ms": avg,
            "median_rtt_ms": med,
            "stddev_rtt_ms": std,
            "jitter_ms": jitter,
            "packet_loss_pct": (self.lost / self.sent * 100.0) if self.sent else 0.0,
            "last_rtt_ms": self.last_rtt,
        }


def probe_tcp(host: str, port: int, timeout: float) -> tuple[bool, Optional[float], Optional[str]]:
    start = time.perf_counter()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            rtt_ms = (time.perf_counter() - start) * 1000.0
            return (True, rtt_ms, None)
    except socket.timeout:
        return (False, None, "timeout")
    except socket.gaierror as err:
        return (False, None, f"dns_error:{err}")
    except OSError as err:
        return (False, None, f"os_error:{err}")


def fmt_ms(value: Optional[float]) -> str:
    if value is None:
        return "-"
    return f"{value:.2f}ms"


def clear_console() -> None:
    sys.stdout.write("\033[2J\033[H")
    sys.stdout.flush()


def render_dashboard(states: list[HostState], start_ts: float, end_ts: Optional[float]) -> None:
    clear_console()
    now = datetime.now(timezone.utc).isoformat()
    print(f"Network Latency Monitor  {now}")
    if end_ts is None:
        print(f"Elapsed: {int(time.time() - start_ts)}s")
    else:
        remaining = max(0, int(end_ts - time.time()))
        print(f"Remaining: {remaining}s")
    print("-" * 90)
    print("Host                      Last RTT   Avg RTT    Min/Max            Loss %   Alert")
    print("-" * 90)
    for st in states:
        s = st.stats()
        host_display = f"{st.label} ({st.host})"
        if len(host_display) > 25:
            host_display = host_display[:22] + "..."
        host_display = host_display.ljust(25)
        minmax = f"{fmt_ms(s['min_rtt_ms'])}/{fmt_ms(s['max_rtt_ms'])}"
        print(
            f"{host_display} {fmt_ms(s['last_rtt_ms']).ljust(10)} {fmt_ms(s['avg_rtt_ms']).ljust(10)} "
            f"{minmax.ljust(18)} {s['packet_loss_pct']:6.2f}%   {'ALERT' if st.last_alert else 'OK'}"
        )
    print("-" * 90)


def should_alert(state: HostState, mode: str, threshold: float, rtt_ms: float) -> bool:
    if mode == "absolute":
        return rtt_ms > threshold
    if not state.rtts:
        return False
    avg = sum(state.rtts) / len(state.rtts)
    return rtt_ms > avg * threshold


def write_json_report(
    output_path: Path,
    states: list[HostState],
    started_at: float,
    ended_at: float,
    interval: float,
    port: int,
    timeout: float,
    duration: Optional[float],
    alert_mode: str,
    alert_value: float,
    interrupted: bool,
) -> None:
    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "monitoring": {
            "started_at": datetime.fromtimestamp(started_at, timezone.utc).isoformat(),
            "ended_at": datetime.fromtimestamp(ended_at, timezone.utc).isoformat(),
            "interrupted": interrupted,
            "interval_sec": interval,
            "port": port,
            "timeout_sec": timeout,
            "duration_sec": duration,
            "alert": {"mode": alert_mode, "value": alert_value},
        },
        "hosts": [],
    }
    for st in states:
        report["hosts"].append(
            {
                "host": st.host,
                "label": st.label,
                "stats": st.stats(),
                "measurements": [
                    {
                        "timestamp": m.timestamp,
                        "success": m.success,
                        "rtt_ms": m.rtt_ms,
                        "error": m.error,
                        "alert": m.alert,
                    }
                    for m in st.measurements
                ],
            }
        )
    output_path.write_text(json.dumps(report, indent=2), encoding="utf-8")


def write_csv(export_path: Path, states: list[HostState]) -> None:
    with export_path.open("w", newline="", encoding="utf-8") as fh:
        writer = csv.writer(fh)
        writer.writerow(["timestamp", "host", "label", "success", "rtt_ms", "error", "alert"])
        for st in states:
            for m in st.measurements:
                writer.writerow([m.timestamp, st.host, st.label, m.success, m.rtt_ms, m.error or "", m.alert])


def final_summary(states: list[HostState]) -> None:
    ranked = sorted(states, key=lambda st: st.stats()["avg_rtt_ms"] if st.stats()["avg_rtt_ms"] is not None else math.inf)
    print("\nFinal Summary (best average RTT first)")
    print("-" * 80)
    for st in ranked:
        s = st.stats()
        print(
            f"{st.label} ({st.host})  avg={fmt_ms(s['avg_rtt_ms'])}  min={fmt_ms(s['min_rtt_ms'])}  "
            f"max={fmt_ms(s['max_rtt_ms'])}  loss={s['packet_loss_pct']:.2f}%  jitter={fmt_ms(s['jitter_ms'])}"
        )


def main() -> int:
    parser = argparse.ArgumentParser(description="Network Latency Monitor")
    parser.add_argument("hosts", nargs="*", help="Hosts to monitor")
    parser.add_argument("--interval", type=float, default=5.0, help="Probe interval in seconds")
    parser.add_argument("--port", type=int, default=80, help="TCP probe port")
    parser.add_argument("--duration", type=str, help="Duration like 60s, 5m, 1h")
    parser.add_argument("--alert", type=str, default="3x", help="Alert threshold (e.g. 3x or 200)")
    parser.add_argument("--output", type=str, default="latency_report.json", help="JSON report path")
    parser.add_argument("--export", type=str, help="CSV export path")
    parser.add_argument("--file", type=str, help="Host file path")
    parser.add_argument("--timeout", type=float, default=5.0, help="Probe timeout in seconds")
    args = parser.parse_args()

    if args.interval <= 0:
        raise ValueError("--interval must be > 0")
    if args.port < 1 or args.port > 65535:
        raise ValueError("--port must be 1..65535")
    if args.timeout <= 0:
        raise ValueError("--timeout must be > 0")

    duration_sec = parse_duration(args.duration)
    alert_mode, alert_value = parse_alert(args.alert)
    host_entries = [(h, h) for h in args.hosts]
    if args.file:
        host_entries.extend(parse_hosts_file(Path(args.file)))

    if not host_entries:
        host_entries = [
            ("8.8.8.8", "Google DNS"),
            ("1.1.1.1", "Cloudflare DNS"),
            ("example.com", "Example"),
            ("google.com", "Google"),
        ]
        duration_sec = 30.0

    states = [HostState(host=h, label=l) for h, l in host_entries]

    stop_flag = {"stop": False}

    def on_sigint(signum, frame):  # type: ignore[no-untyped-def]
        stop_flag["stop"] = True

    signal.signal(signal.SIGINT, on_sigint)

    started_at = time.time()
    end_at = started_at + duration_sec if duration_sec is not None else None
    interrupted = False

    while True:
        if stop_flag["stop"]:
            interrupted = True
            break
        if end_at is not None and time.time() >= end_at:
            break

        tick_start = time.time()
        threads: list[threading.Thread] = []
        results: list[tuple[bool, Optional[float], Optional[str]]] = [(False, None, "not_started")] * len(states)

        def run_probe(idx: int, host: str) -> None:
            results[idx] = probe_tcp(host, args.port, args.timeout)

        for idx, st in enumerate(states):
            t = threading.Thread(target=run_probe, args=(idx, st.host), daemon=True)
            t.start()
            threads.append(t)
        for t in threads:
            t.join()

        for idx, st in enumerate(states):
            ok, rtt_ms, err = results[idx]
            st.sent += 1
            ts = datetime.now(timezone.utc).isoformat()
            if not ok:
                st.lost += 1
                st.last_alert = False
                st.measurements.append(Measurement(timestamp=ts, success=False, rtt_ms=None, error=err, alert=False))
                continue
            assert rtt_ms is not None
            is_alert = should_alert(st, alert_mode, alert_value, rtt_ms)
            if st.last_rtt is not None:
                st.jitters.append(abs(rtt_ms - st.last_rtt))
            st.last_rtt = rtt_ms
            st.rtts.append(rtt_ms)
            st.last_alert = is_alert
            st.measurements.append(Measurement(timestamp=ts, success=True, rtt_ms=rtt_ms, error=None, alert=is_alert))

        render_dashboard(states, started_at, end_at)
        elapsed = time.time() - tick_start
        sleep_for = max(0.0, args.interval - elapsed)
        time.sleep(sleep_for)

    ended_at = time.time()
    render_dashboard(states, started_at, end_at)
    final_summary(states)

    output_path = Path(args.output)
    write_json_report(
        output_path=output_path,
        states=states,
        started_at=started_at,
        ended_at=ended_at,
        interval=args.interval,
        port=args.port,
        timeout=args.timeout,
        duration=duration_sec,
        alert_mode=alert_mode,
        alert_value=alert_value,
        interrupted=interrupted,
    )
    print(f"\nSaved JSON report: {output_path}")

    if args.export:
        export_path = Path(args.export)
        write_csv(export_path, states)
        print(f"Saved CSV export: {export_path}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())