Network QoS Analyser with Claude AI

Collect live network metrics and use Claude vision API to detect congestion.

Introduction

30 Second Summary

You know when your video call freezes or your game starts lagging? Something is going wrong on your network, but figuring out exactly what is happening (and when) usually requires expensive monitoring tools or guesswork.

In this project, you will build a Python tool that collects real-time network performance data from your own connection, turns it into visual graphs, and sends those graphs to Claude AI for expert-level congestion analysis.

What You'll Build

A Python-based Network QoS Graphs Analyser that pings live hosts on your network, visualizes the collected metrics as time-series graphs, and uses Claude's vision API to interpret those graphs for congestion patterns and preventive recommendations.

By the end of this project, you'll have:

  • A data collector script that pings real hosts and records latency, jitter, and packet loss to a CSV file.
  • A graph generator that produces a QoS dashboard image with threshold lines and anomaly highlighting.
  • An AI analyser that sends your dashboard to Claude's vision API and prints a detailed congestion analysis report.
  • Secret Mission: Add a continuous monitoring mode that compares rolling data windows and detects deterioration trends over time.

Are there any prerequisites?

None. This guide walks you through installing every tool you need, including Python, your code editor, and the Anthropic API key. No prior networking or AI experience is required.

Build the Real-Time QoS Data Collector

Your environment is ready with Python, your dependencies, and your API key all verified. Now it is time to build the first stage of your pipeline: collecting live network data.

Network monitoring starts with measuring how packets travel across the internet. You will write a script that pings real servers on the internet and records three key metrics: latency (how long a round-trip takes), jitter (how much latency varies between packets), and packet loss (how many packets never arrive).

In this step, get ready to:
  • Build a function that pings hosts and parses network performance data.
  • Write a collection loop that gathers samples over time and saves them to CSV.
  • Run the collector against real servers and verify your data file.
Create the ping_host() function

The ping command sends small packets to a server and measures how long each one takes to return. Your function will run this command using Python's subprocess module, then parse the output with regex to extract the timing data.

  • In Cursor, create a new file called collect_qos.py inside your qos-analyser/ folder.
  • Add the imports and the ping_host() function by pasting this code:
import csv
import subprocess
import re
import time
import sys
from datetime import datetime


def ping_host(host, count=4):
    """Ping a host and return RTT values and packet loss."""
    try:
        if sys.platform == "win32":
            cmd = ["ping", "-n", str(count), host]
        else:
            cmd = ["ping", "-c", str(count), host]

        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        output = result.stdout

        # Extract RTT values
        if sys.platform == "win32":
            rtt_pattern = r"time[=<](\d+)ms"
        else:
            rtt_pattern = r"time[=](\d+\.?\d*)\s*ms"

        rtts = [float(x) for x in re.findall(rtt_pattern, output)]

        # Extract packet loss
        loss_pattern = r"(\d+)%\s*(packet)?\s*loss"
        loss_match = re.search(loss_pattern, output)
        packet_loss = float(loss_match.group(1)) if loss_match else 0.0

        return rtts, packet_loss

    except (subprocess.TimeoutExpired, FileNotFoundError, Exception) as e:
        print(f"  Warning: ping to {host} failed: {e}")
        return [], 100.0

What does this code do?

The function detects your operating system to use the correct ping flag. Windows uses -n for the packet count while macOS and Linux use -c.

After running the ping, two regex patterns extract the data. The first pattern pulls individual RTT (round-trip time) values in milliseconds. The second pattern finds the packet loss percentage from the summary line.

If the ping fails entirely (network down, host unreachable, timeout), the function returns an empty list and 100% loss instead of crashing your script.

Why subprocess instead of a Python ping library?

Third-party Python ping libraries often require root or administrator access because they create raw ICMP sockets. The system ping command already has the necessary permissions on every operating system, so your script works without elevated privileges.

  • Save collect_qos.py.

Seeing a red underline on an import?

Make sure your virtual environment is active in Cursor. Check the bottom status bar for the Python interpreter path pointing to your venv folder.

help me fix my Python interpreter in Cursor

Write the collection loop

A single ping gives you one snapshot. To spot patterns, you need many samples over time. The collection loop will ping three public DNS servers every 2 seconds and record the results to a CSV file.

  • Below the ping_host() function in collect_qos.py, add the calculate_jitter() helper function:
def calculate_jitter(rtts):
    """Calculate jitter as mean absolute difference between consecutive RTTs."""
    if len(rtts) < 2:
        return 0.0
    differences = [abs(rtts[i] - rtts[i - 1]) for i in range(1, len(rtts))]
    return sum(differences) / len(differences)

What is jitter?

Jitter measures how inconsistent your latency is. If one packet takes 20ms and the next takes 80ms, that 60ms difference is jitter.

The function calculates jitter by averaging the absolute differences between consecutive RTT values. High jitter causes choppy video calls and laggy online games even when average latency looks fine.

  • Below calculate_jitter(), add the main collect_samples() function:
def collect_samples(hosts, duration_seconds=120, interval_seconds=2, output_file="qos_data.csv"):
    """Collect QoS samples by pinging hosts at regular intervals."""
    print(f"Collecting QoS data for {duration_seconds}s (interval: {interval_seconds}s)")
    print(f"  Hosts: {', '.join(hosts)}")

    fieldnames = ["timestamp", "host", "latency_ms", "jitter_ms", "packet_loss_percent"]
    samples_collected = 0

    with open(output_file, "w", newline="") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        start_time = time.time()
        while (time.time() - start_time) < duration_seconds:
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            for host in hosts:
                rtts, packet_loss = ping_host(host, count=4)

                if rtts:
                    latency = sum(rtts) / len(rtts)
                    jitter = calculate_jitter(rtts)
                else:
                    latency = 0.0
                    jitter = 0.0

                writer.writerow({
                    "timestamp": timestamp,
                    "host": host,
                    "latency_ms": round(latency, 2),
                    "jitter_ms": round(jitter, 2),
                    "packet_loss_percent": round(packet_loss, 1),
                })
                samples_collected += 1

            elapsed = time.time() - start_time
            remaining = duration_seconds - elapsed
            print(f"  [{elapsed:.0f}s/{duration_seconds}s] Collected {samples_collected} samples", end="\r")

            if remaining > interval_seconds:
                time.sleep(interval_seconds)

    print(f"\n  Done! {samples_collected} samples saved to '{output_file}'")
    return output_file

What does this function do?

The function runs a timed loop that pings each host every 2 seconds. For each ping round, it calls ping_host() to get raw RTT values, then computes the average latency and jitter.

Results go straight into a CSV file using Python's csv.DictWriter. The progress counter updates on a single line using end="\r" so you can watch the collection happen in real time.

  • Save collect_qos.py.

Worried about the 2-minute collection time?

The default duration of 120 seconds is intentional. You need enough samples to produce meaningful graphs in the next step. Let it run while you read ahead.

help me understand the collection timing

Run the collector and verify your data

The script needs an entry point so you can run it directly from the terminal. The if __name__ == "__main__" block ensures the collection only starts when you run the file on its own.

  • At the bottom of collect_qos.py, add the entry point:
if __name__ == "__main__":
    hosts = ["8.8.8.8", "1.1.1.1", "8.8.4.4"]
    collect_samples(hosts, duration_seconds=120, interval_seconds=2)

What are these hosts?

These are public DNS servers run by Google (8.8.8.8, 8.8.4.4) and Cloudflare (1.1.1.1). They are designed to be always reachable and respond to pings from anywhere in the world.

  • Save collect_qos.py.
  • Run the collector by executing this command in your terminal:
python collect_qos.py

What should I see?

The script prints the hosts it will ping, then shows a live progress counter updating every few seconds. It will run for about 2 minutes. When it finishes, you will see a summary like:

Done! 180 samples saved to 'qos_data.csv'

The exact sample count depends on your network speed. You should see roughly 180 samples (3 hosts times 60 ping rounds).

  • After the script finishes, verify the CSV file exists by running:
head -5 qos_data.csv

You should see the CSV header row followed by rows of real network data with timestamps, host addresses, and metric values.

All latency values showing 0.0?

This usually means the regex did not match your system's ping output format. Some locales or older systems use slightly different formatting.

  • Try running ping -c 4 8.8.8.8 (or ping -n 4 8.8.8.8 on Windows) manually and check that the output contains time= followed by a number.
  • If your network blocks ICMP packets, try a different network or use a mobile hotspot.

help me debug my ping output parsing

✔️ Awesome, I've got everything!

Great. Make sure you have saved collect_qos.py and confirmed that qos_data.csv contains real data.

ⓧ I'd like to double check the full code

import csv
import subprocess
import re
import time
import sys
from datetime import datetime


def ping_host(host, count=4):
    """Ping a host and return RTT values and packet loss."""
    try:
        if sys.platform == "win32":
            cmd = ["ping", "-n", str(count), host]
        else:
            cmd = ["ping", "-c", str(count), host]

        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        output = result.stdout

        # Extract RTT values
        if sys.platform == "win32":
            rtt_pattern = r"time[=<](\d+)ms"
        else:
            rtt_pattern = r"time[=](\d+\.?\d*)\s*ms"

        rtts = [float(x) for x in re.findall(rtt_pattern, output)]

        # Extract packet loss
        loss_pattern = r"(\d+)%\s*(packet)?\s*loss"
        loss_match = re.search(loss_pattern, output)
        packet_loss = float(loss_match.group(1)) if loss_match else 0.0

        return rtts, packet_loss

    except (subprocess.TimeoutExpired, FileNotFoundError, Exception) as e:
        print(f"  Warning: ping to {host} failed: {e}")
        return [], 100.0


def calculate_jitter(rtts):
    """Calculate jitter as mean absolute difference between consecutive RTTs."""
    if len(rtts) < 2:
        return 0.0
    differences = [abs(rtts[i] - rtts[i - 1]) for i in range(1, len(rtts))]
    return sum(differences) / len(differences)


def collect_samples(hosts, duration_seconds=120, interval_seconds=2, output_file="qos_data.csv"):
    """Collect QoS samples by pinging hosts at regular intervals."""
    print(f"Collecting QoS data for {duration_seconds}s (interval: {interval_seconds}s)")
    print(f"  Hosts: {', '.join(hosts)}")

    fieldnames = ["timestamp", "host", "latency_ms", "jitter_ms", "packet_loss_percent"]
    samples_collected = 0

    with open(output_file, "w", newline="") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        start_time = time.time()
        while (time.time() - start_time) < duration_seconds:
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

            for host in hosts:
                rtts, packet_loss = ping_host(host, count=4)

                if rtts:
                    latency = sum(rtts) / len(rtts)
                    jitter = calculate_jitter(rtts)
                else:
                    latency = 0.0
                    jitter = 0.0

                writer.writerow({
                    "timestamp": timestamp,
                    "host": host,
                    "latency_ms": round(latency, 2),
                    "jitter_ms": round(jitter, 2),
                    "packet_loss_percent": round(packet_loss, 1),
                })
                samples_collected += 1

            elapsed = time.time() - start_time
            remaining = duration_seconds - elapsed
            print(f"  [{elapsed:.0f}s/{duration_seconds}s] Collected {samples_collected} samples", end="\r")

            if remaining > interval_seconds:
                time.sleep(interval_seconds)

    print(f"\n  Done! {samples_collected} samples saved to '{output_file}'")
    return output_file


if __name__ == "__main__":
    hosts = ["8.8.8.8", "1.1.1.1", "8.8.4.4"]
    collect_samples(hosts, duration_seconds=120, interval_seconds=2)

Your data collector is working and you have real network measurements saved to CSV. Next up, you will turn that raw data into a visual dashboard with graphs that make patterns easy to spot.

Generate the QoS Dashboard Graph

Your collector script has filled qos_data.csv with real network measurements. Numbers in a spreadsheet are useful, but they don't reveal patterns the way a visual graph does.

In this step, you will build a graph generator that reads the CSV and produces a three-panel dashboard image. Each panel tracks a different QoS metric over time with threshold lines that highlight when performance degrades. This image is exactly what you will send to Claude for AI analysis in the next step.

In this step, get ready to:
  • Load and aggregate CSV data into a time-series DataFrame.
  • Create a three-panel matplotlib dashboard with threshold lines and anomaly highlighting.
  • Save the dashboard as a PNG and verify it visually.
Create the data loading function

The CSV contains one row per host per timestamp. To graph a single time series, you need to average across all three hosts at each timestamp. The pandas library makes this grouping and aggregation straightforward.

  • Create a new file called graph_qos.py in your qos-analyser/ folder.
  • Add the imports, threshold constants, and the load_and_aggregate() function by pasting this code:
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates


LATENCY_THRESHOLD = 100  # ms
JITTER_THRESHOLD = 30  # ms
PACKET_LOSS_THRESHOLD = 1  # percent


def load_and_aggregate(filepath="qos_data.csv"):
    """Load collected data and aggregate across hosts per timestamp."""
    df = pd.read_csv(filepath, parse_dates=["timestamp"])
    aggregated = df.groupby("timestamp").agg({
        "latency_ms": "mean",
        "jitter_ms": "mean",
        "packet_loss_percent": "mean",
    }).reset_index()
    aggregated.set_index("timestamp", inplace=True)
    aggregated.sort_index(inplace=True)
    return aggregated

What does this code do?

The three constants at the top define industry-standard QoS thresholds. Any metric exceeding these values signals potential congestion.

  • LATENCY_THRESHOLD = 100 means latency above 100ms is concerning for most connections.
  • pd.read_csv(filepath, parse_dates=["timestamp"]) reads the CSV and converts the timestamp column into proper datetime objects for plotting.
  • groupby("timestamp").agg({..."mean"}) averages the three hosts' values at each timestamp into a single data point per time slot.
  • Save graph_qos.py.

You won't see a change yet. The next chunk adds the statistics helper, then the dashboard function brings everything together visually.

Next, you need a way to summarize the collected data into key statistics. These numbers will appear in the terminal output and later get passed to Claude as text context alongside the graph.

  • Add the compute_statistics() function below load_and_aggregate() in graph_qos.py:
def compute_statistics(df):
    """Compute summary statistics for the report."""
    stats = {}
    for col in df.columns:
        stats[col] = {
            "min": df[col].min(),
            "max": df[col].max(),
            "mean": df[col].mean(),
            "p95": df[col].quantile(0.95),
        }
    return stats

What does this code do?

This function loops through each metric column and computes four summary values: minimum, maximum, mean, and the 95th percentile (p95). The p95 value tells you what the "worst normal" performance looks like by excluding the top 5% of outliers.

  • Save graph_qos.py.

You won't see a change yet. The next chunk builds the dashboard visualization that uses both of these functions.

Write the dashboard plotting function

The dashboard uses matplotlib to create three stacked time-series panels. Each panel plots one metric, draws a red threshold line, and shades the regions where the value exceeds the threshold. This makes congestion periods immediately visible.

  • Add the create_dashboard() function below compute_statistics() in graph_qos.py:
def create_dashboard(df, output_file="qos_dashboard.png"):
    """Create a 3-panel QoS dashboard and save as PNG."""
    fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)
    fig.suptitle("Network QoS Dashboard - Real-Time Collection", fontsize=14, fontweight="bold")

    # Latency
    ax = axes[0]
    ax.plot(df.index, df["latency_ms"], linewidth=1.0, color="#2196F3", label="Latency")
    ax.axhline(y=LATENCY_THRESHOLD, color="red", linestyle="--", linewidth=1.2, label=f"Threshold ({LATENCY_THRESHOLD} ms)")
    mask = df["latency_ms"] > LATENCY_THRESHOLD
    ax.fill_between(df.index, 0, df["latency_ms"].max() * 1.1, where=mask, alpha=0.1, color="red")
    ax.set_ylabel("Latency (ms)")
    ax.legend(loc="upper right", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Jitter
    ax = axes[1]
    ax.plot(df.index, df["jitter_ms"], linewidth=1.0, color="#FF9800", label="Jitter")
    ax.axhline(y=JITTER_THRESHOLD, color="red", linestyle="--", linewidth=1.2, label=f"Threshold ({JITTER_THRESHOLD} ms)")
    mask = df["jitter_ms"] > JITTER_THRESHOLD
    ax.fill_between(df.index, 0, df["jitter_ms"].max() * 1.1, where=mask, alpha=0.1, color="red")
    ax.set_ylabel("Jitter (ms)")
    ax.legend(loc="upper right", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Packet Loss
    ax = axes[2]
    ax.plot(df.index, df["packet_loss_percent"], linewidth=1.0, color="#4CAF50", label="Packet Loss")
    ax.axhline(y=PACKET_LOSS_THRESHOLD, color="red", linestyle="--", linewidth=1.2, label=f"Threshold ({PACKET_LOSS_THRESHOLD}%)")
    mask = df["packet_loss_percent"] > PACKET_LOSS_THRESHOLD
    ax.fill_between(df.index, 0, df["packet_loss_percent"].max() * 1.1 + 0.5, where=mask, alpha=0.1, color="red")
    ax.set_ylabel("Packet Loss (%)")
    ax.set_xlabel("Time")
    ax.legend(loc="upper right", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Format x-axis
    for a in axes:
        a.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))

    # Add collection info
    time_range = f"{df.index[0].strftime('%H:%M:%S')} - {df.index[-1].strftime('%H:%M:%S')}"
    fig.text(0.5, 0.01, f"Collection period: {time_range} | Samples: {len(df)}", ha="center", fontsize=9, style="italic")

    plt.tight_layout(rect=[0, 0.03, 1, 0.97])
    plt.savefig(output_file, dpi=150)
    plt.close()
    print(f"  Dashboard saved: {output_file}")
    return output_file

What does this code do?

This function builds the visual dashboard that Claude will analyse in the next step.

  • plt.subplots(3, 1, figsize=(12, 10), sharex=True) creates three vertically stacked panels sharing the same time axis.
  • axhline() draws a red dashed horizontal line at each threshold value so you can instantly see when performance is poor.
  • fill_between() shades the time regions where values exceed the threshold in transparent red. This highlights congestion periods visually.
  • fig.text() adds an annotation at the bottom showing the collection time range and sample count. This gives Claude extra context when reading the image.
  • Save graph_qos.py.

You won't see a graph yet. The next chunk adds the entry point that calls these functions so you can run the script.

Run the graph generator and verify the output

With all three functions in place, you need an entry point that loads the data, prints statistics, and generates the dashboard image.

  • Add this __main__ block at the bottom of graph_qos.py:
if __name__ == "__main__":
    df = load_and_aggregate()
    stats = compute_statistics(df)
    print("QoS Statistics:")
    for metric, values in stats.items():
        print(f"  {metric}: mean={values['mean']:.2f}, max={values['max']:.2f}, p95={values['p95']:.2f}")
    create_dashboard(df)

What does this code do?

This entry point runs the full graphing pipeline: load the CSV, compute statistics, print them to the terminal, and generate the dashboard PNG. It only executes when you run graph_qos.py directly (not when another script imports from it).

  • Save graph_qos.py.
  • Run the graph generator by running this command in your terminal:
python graph_qos.py

You should see statistics printed in the terminal, followed by a line confirming the dashboard was saved.

What should I see?

Your terminal output will look something like this (your numbers will differ based on your network):

QoS Statistics:
  latency_ms: mean=12.45, max=34.21, p95=28.67
  jitter_ms: mean=3.12, max=15.40, p95=10.55
  packet_loss_percent: mean=0.00, max=0.00, p95=0.00
  Dashboard saved: qos_dashboard.png
  • Open qos_dashboard.png in your file browser or image viewer to verify it looks correct.

You should see three stacked panels. The top panel shows latency in blue, the middle shows jitter in orange, and the bottom shows packet loss in green. Each panel has a red dashed threshold line.

Seeing an error or a blank image?

  • If you see a FileNotFoundError mentioning qos_data.csv, make sure you are running the command from inside the qos-analyser/ folder where the CSV was saved in Step 2.
  • If you see an ImportError for pandas or matplotlib, confirm your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
  • If the image file is created but looks empty, check that qos_data.csv has actual data rows (not just a header). Re-run collect_qos.py from Step 2 if needed.

Help me debug my graph_qos.py script

✔️ Awesome, I've got everything!

Great. Double check you have saved graph_qos.py and confirmed the dashboard image looks correct.

ⓧ I'd like to double check the full code

Here is the complete graph_qos.py file for comparison:

import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates


LATENCY_THRESHOLD = 100  # ms
JITTER_THRESHOLD = 30  # ms
PACKET_LOSS_THRESHOLD = 1  # percent


def load_and_aggregate(filepath="qos_data.csv"):
    """Load collected data and aggregate across hosts per timestamp."""
    df = pd.read_csv(filepath, parse_dates=["timestamp"])
    aggregated = df.groupby("timestamp").agg({
        "latency_ms": "mean",
        "jitter_ms": "mean",
        "packet_loss_percent": "mean",
    }).reset_index()
    aggregated.set_index("timestamp", inplace=True)
    aggregated.sort_index(inplace=True)
    return aggregated


def compute_statistics(df):
    """Compute summary statistics for the report."""
    stats = {}
    for col in df.columns:
        stats[col] = {
            "min": df[col].min(),
            "max": df[col].max(),
            "mean": df[col].mean(),
            "p95": df[col].quantile(0.95),
        }
    return stats


def create_dashboard(df, output_file="qos_dashboard.png"):
    """Create a 3-panel QoS dashboard and save as PNG."""
    fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)
    fig.suptitle("Network QoS Dashboard - Real-Time Collection", fontsize=14, fontweight="bold")

    # Latency
    ax = axes[0]
    ax.plot(df.index, df["latency_ms"], linewidth=1.0, color="#2196F3", label="Latency")
    ax.axhline(y=LATENCY_THRESHOLD, color="red", linestyle="--", linewidth=1.2, label=f"Threshold ({LATENCY_THRESHOLD} ms)")
    mask = df["latency_ms"] > LATENCY_THRESHOLD
    ax.fill_between(df.index, 0, df["latency_ms"].max() * 1.1, where=mask, alpha=0.1, color="red")
    ax.set_ylabel("Latency (ms)")
    ax.legend(loc="upper right", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Jitter
    ax = axes[1]
    ax.plot(df.index, df["jitter_ms"], linewidth=1.0, color="#FF9800", label="Jitter")
    ax.axhline(y=JITTER_THRESHOLD, color="red", linestyle="--", linewidth=1.2, label=f"Threshold ({JITTER_THRESHOLD} ms)")
    mask = df["jitter_ms"] > JITTER_THRESHOLD
    ax.fill_between(df.index, 0, df["jitter_ms"].max() * 1.1, where=mask, alpha=0.1, color="red")
    ax.set_ylabel("Jitter (ms)")
    ax.legend(loc="upper right", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Packet Loss
    ax = axes[2]
    ax.plot(df.index, df["packet_loss_percent"], linewidth=1.0, color="#4CAF50", label="Packet Loss")
    ax.axhline(y=PACKET_LOSS_THRESHOLD, color="red", linestyle="--", linewidth=1.2, label=f"Threshold ({PACKET_LOSS_THRESHOLD}%)")
    mask = df["packet_loss_percent"] > PACKET_LOSS_THRESHOLD
    ax.fill_between(df.index, 0, df["packet_loss_percent"].max() * 1.1 + 0.5, where=mask, alpha=0.1, color="red")
    ax.set_ylabel("Packet Loss (%)")
    ax.set_xlabel("Time")
    ax.legend(loc="upper right", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Format x-axis
    for a in axes:
        a.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))

    # Add collection info
    time_range = f"{df.index[0].strftime('%H:%M:%S')} - {df.index[-1].strftime('%H:%M:%S')}"
    fig.text(0.5, 0.01, f"Collection period: {time_range} | Samples: {len(df)}", ha="center", fontsize=9, style="italic")

    plt.tight_layout(rect=[0, 0.03, 1, 0.97])
    plt.savefig(output_file, dpi=150)
    plt.close()
    print(f"  Dashboard saved: {output_file}")
    return output_file


if __name__ == "__main__":
    df = load_and_aggregate()
    stats = compute_statistics(df)
    print("QoS Statistics:")
    for metric, values in stats.items():
        print(f"  {metric}: mean={values['mean']:.2f}, max={values['max']:.2f}, p95={values['p95']:.2f}")
    create_dashboard(df)

Your dashboard is ready. You now have a professional-looking visualization of your network's health over time. Next up, you will send this image to Claude and let AI analyse it for congestion patterns that might not be obvious from the numbers alone.

Send the Graph to Claude for Congestion Analysis

You have real network data in a CSV and a polished dashboard graph. Numbers and visuals are useful, but interpreting QoS graphs for congestion patterns takes expertise.

In this step, you will send the dashboard image directly to Claude's vision API. Claude will analyse the graph like a network engineer would, spotting congestion patterns, correlating metrics across the three panels, and recommending preventive actions.

In this step, get ready to:
  • Build an image encoder that prepares the dashboard for the Claude API.
  • Craft a multi-modal API call with the graph image and raw statistics as context.
  • Run the analyser and review Claude's congestion report.
Create the image encoder

Claude's Messages API accepts images as base64-encoded strings. You need a helper function that reads the PNG file from disk and converts it to this format.

  • Create a new file called analyse_with_claude.py in your qos-analyser folder.
  • Add the imports and the encode_image() function by pasting this code:
import base64
import anthropic
from graph_qos import load_and_aggregate, compute_statistics


def encode_image(filepath):
    """Read an image file and return its base64 encoding."""
    with open(filepath, "rb") as f:
        return base64.standard_b64encode(f.read()).decode("utf-8")

What does this code do?

  • The imports pull in base64 for encoding, the anthropic SDK for calling Claude, and two helper functions from your graph_qos.py (to load data and compute stats).
  • encode_image() opens the PNG in binary mode, encodes the raw bytes to a base64 string, and returns it. This is the format Claude's vision API expects for inline images.
  • Save analyse_with_claude.py.

You won't see output yet. The next chunk adds the function that actually calls Claude.

Build the Claude API call with image and statistics

A graph alone gives Claude visual context, but pairing it with raw numbers lets Claude cross-reference what it sees with precise values. You will build a statistics formatter and the main analysis function.

  • In analyse_with_claude.py, add the build_statistics_summary() function below your encode_image() function:
def build_statistics_summary(stats):
    """Format statistics into a text summary for Claude."""
    lines = ["QoS Statistics Summary:"]
    thresholds = {
        "latency_ms": ("Latency", "< 100 ms"),
        "jitter_ms": ("Jitter", "< 30 ms"),
        "packet_loss_percent": ("Packet Loss", "< 1%"),
    }
    for col, (name, target) in thresholds.items():
        s = stats[col]
        lines.append(f"  {name}: mean={s['mean']:.2f}, max={s['max']:.2f}, p95={s['p95']:.2f} (target: {target})")
    return "\n".join(lines)

What does this code do?

build_statistics_summary() takes the stats dictionary (from compute_statistics() in graph_qos.py) and formats it into a readable text block. Each metric gets its mean, max, and 95th percentile values alongside the acceptable threshold. This gives Claude precise numbers to reference when interpreting the visual patterns.

  • Save analyse_with_claude.py.

You won't see output yet. The next chunk builds the main function that ties everything together.

  • Still in analyse_with_claude.py, add the analyse_dashboard() function and the if __name__ block below build_statistics_summary():
def analyse_dashboard(image_path="qos_dashboard.png", data_path="qos_data.csv"):
    """Send the QoS dashboard to Claude for congestion analysis."""
    print("  Sending dashboard to Claude for analysis...")

    # Load statistics for context
    df = load_and_aggregate(data_path)
    stats = compute_statistics(df)
    stats_summary = build_statistics_summary(stats)

    # Encode the dashboard image
    image_data = encode_image(image_path)

    # Create the Claude client
    client = anthropic.Anthropic()

    # Send to Claude with vision
    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1500,
        system="You are an expert network engineer analysing QoS (Quality of Service) graphs to identify network congestion. Analyse the provided graph image along with the statistics. Focus on: 1) Identifying congestion periods visible in the data, 2) Explaining what network conditions the patterns suggest, 3) Recommending specific preventive actions. Be concise and actionable.",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": f"Analyse this network QoS dashboard for congestion indicators. Here are the raw statistics:\n\n{stats_summary}\n\nIdentify any congestion or pre-congestion patterns. What do the graphs tell you about the network health? What specific actions should be taken to prevent future congestion?",
                    },
                ],
            }
        ],
    )

    # Extract and display the analysis
    analysis_text = message.content[0].text

    print("\n" + "=" * 60)
    print("  CLAUDE AI CONGESTION ANALYSIS")
    print("=" * 60)
    print(analysis_text)
    print("=" * 60)
    print(f"\n  Tokens used: {message.usage.input_tokens} input, {message.usage.output_tokens} output")

    return analysis_text


if __name__ == "__main__":
    analyse_dashboard()

What does this code do?

  • analyse_dashboard() loads your CSV stats and encodes the dashboard PNG, then sends both to Claude in a single API call.
  • anthropic.Anthropic() creates a client that automatically reads your ANTHROPIC_API_KEY environment variable.
  • The system parameter tells Claude to behave as a network engineer focused on congestion detection.
  • The messages array sends a user message containing two content blocks: the base64 image and a text prompt with the raw statistics. This multi-modal input gives Claude both the visual graph and precise numbers.
  • The response's message.usage shows token counts so you can estimate the API cost of each analysis.

Why send the image instead of just numbers?

A network engineer looking at a graph spots patterns that raw numbers miss: gradual ramps, correlated spikes across metrics, periodic oscillations. By sending the actual graph, Claude can identify these visual patterns the same way a human expert would. The statistics text provides precision to back up what Claude sees.

  • Save analyse_with_claude.py.

Seeing a syntax error when you save?

  • Check that the messages.create() call has matching brackets and commas. The nested dictionary structure is deep here.
  • Make sure the f-string inside the text field uses \n\n (escaped newlines), not actual line breaks inside the string.
  • help me fix a syntax error in my analyse_with_claude.py file

✔️ Awesome, I've got everything!

Great. Make sure you have saved analyse_with_claude.py before running it in the next substep.

ⓧ I'd like to double check the full code

import base64
import anthropic
from graph_qos import load_and_aggregate, compute_statistics


def encode_image(filepath):
    """Read an image file and return its base64 encoding."""
    with open(filepath, "rb") as f:
        return base64.standard_b64encode(f.read()).decode("utf-8")


def build_statistics_summary(stats):
    """Format statistics into a text summary for Claude."""
    lines = ["QoS Statistics Summary:"]
    thresholds = {
        "latency_ms": ("Latency", "< 100 ms"),
        "jitter_ms": ("Jitter", "< 30 ms"),
        "packet_loss_percent": ("Packet Loss", "< 1%"),
    }
    for col, (name, target) in thresholds.items():
        s = stats[col]
        lines.append(f"  {name}: mean={s['mean']:.2f}, max={s['max']:.2f}, p95={s['p95']:.2f} (target: {target})")
    return "\n".join(lines)


def analyse_dashboard(image_path="qos_dashboard.png", data_path="qos_data.csv"):
    """Send the QoS dashboard to Claude for congestion analysis."""
    print("  Sending dashboard to Claude for analysis...")

    # Load statistics for context
    df = load_and_aggregate(data_path)
    stats = compute_statistics(df)
    stats_summary = build_statistics_summary(stats)

    # Encode the dashboard image
    image_data = encode_image(image_path)

    # Create the Claude client
    client = anthropic.Anthropic()

    # Send to Claude with vision
    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1500,
        system="You are an expert network engineer analysing QoS (Quality of Service) graphs to identify network congestion. Analyse the provided graph image along with the statistics. Focus on: 1) Identifying congestion periods visible in the data, 2) Explaining what network conditions the patterns suggest, 3) Recommending specific preventive actions. Be concise and actionable.",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": f"Analyse this network QoS dashboard for congestion indicators. Here are the raw statistics:\n\n{stats_summary}\n\nIdentify any congestion or pre-congestion patterns. What do the graphs tell you about the network health? What specific actions should be taken to prevent future congestion?",
                    },
                ],
            }
        ],
    )

    # Extract and display the analysis
    analysis_text = message.content[0].text

    print("\n" + "=" * 60)
    print("  CLAUDE AI CONGESTION ANALYSIS")
    print("=" * 60)
    print(analysis_text)
    print("=" * 60)
    print(f"\n  Tokens used: {message.usage.input_tokens} input, {message.usage.output_tokens} output")

    return analysis_text


if __name__ == "__main__":
    analyse_dashboard()
Run the analyser and review Claude's report

Time to see what Claude thinks of your network. The script will read your dashboard, send it to the API, and print a formatted congestion analysis report.

  • Run the analysis script by entering this command in your terminal:
python analyse_with_claude.py
  • Wait a few seconds for the API call to complete. You should see a bordered report headed CLAUDE AI CONGESTION ANALYSIS followed by Claude's interpretation of your network graphs. At the bottom, you will see the token counts for input and output.

What should I see?

Claude's report will describe what it sees in each panel of your dashboard. It identifies periods where metrics crossed thresholds, explains what those patterns suggest about your network conditions, and recommends specific actions.

The token usage line shows how many tokens the image and text consumed. A typical run uses around 1500-2500 input tokens (mostly from the image) and 300-800 output tokens. At claude-haiku-4-5 pricing ($1/MTok input, $5/MTok output), each analysis costs well under $0.01.

Seeing an authentication error?

  • Make sure your ANTHROPIC_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you need to export it again.
  • Verify the key is correct by checking it against the value on platform.claude.com.
  • If you see a file-not-found error, confirm that qos_dashboard.png and qos_data.csv exist in your current directory (you generated them in the previous steps).
  • help me debug an error when running analyse_with_claude.py

Your analyser is working. Claude just interpreted your real network data like a network engineer would, spotting patterns and giving actionable advice. Next up, you will wire all three scripts together into a single pipeline that runs the full workflow with one command.

Orchestrate the Full Pipeline

You now have three scripts that each handle one stage of the analysis: collecting data, generating graphs, and sending graphs to Claude. Running them manually in sequence works, but a single orchestrator script makes the whole pipeline repeatable with one command.

In this step, you will create main.py to tie all three stages together. It will time each stage, print progress, and accept command-line arguments so you can adjust the collection without editing code.

In this step, get ready to:
  • Create the pipeline orchestrator that runs collect, graph, and analyse in sequence.
  • Add command-line arguments for duration, interval, and hosts.
  • Run the full pipeline end-to-end and review Claude's analysis of your network.
Create the pipeline orchestrator

The orchestrator imports functions from your three existing scripts and calls them in order. It also records how long each stage takes so you can see where time is spent.

  • Create a new file called main.py in your qos-analyser/ folder.
  • Add the imports and run_pipeline() function by pasting this code:
import argparse
import time

from collect_qos import collect_samples
from graph_qos import load_and_aggregate, compute_statistics, create_dashboard
from analyse_with_claude import analyse_dashboard


def run_pipeline(duration, interval, hosts):
    """Run the full QoS analysis pipeline."""
    print("=" * 60)
    print("  NETWORK QoS GRAPHS ANALYSER (with Claude AI)")
    print("=" * 60)

    # Stage 1: Collect data
    print("\n[Stage 1] Collecting real-time QoS data...")
    stage_start = time.time()
    data_file = collect_samples(hosts, duration_seconds=duration, interval_seconds=interval)
    print(f"  Stage 1 completed in {time.time() - stage_start:.1f}s")

    # Stage 2: Generate graphs
    print("\n[Stage 2] Generating QoS dashboard...")
    stage_start = time.time()
    df = load_and_aggregate(data_file)
    stats = compute_statistics(df)
    print("  Statistics:")
    for metric, values in stats.items():
        print(f"    {metric}: mean={values['mean']:.2f}, max={values['max']:.2f}, p95={values['p95']:.2f}")
    dashboard_file = create_dashboard(df)
    print(f"  Stage 2 completed in {time.time() - stage_start:.1f}s")

    # Stage 3: AI analysis
    print("\n[Stage 3] Analysing with Claude AI...")
    stage_start = time.time()
    analysis = analyse_dashboard(dashboard_file, data_file)
    print(f"  Stage 3 completed in {time.time() - stage_start:.1f}s")

    print("\nPipeline complete!")
    return analysis

What does this code do?

The run_pipeline() function runs your three scripts in the correct order. Each stage records a start time with time.time() and prints how long it took when it finishes.

  • Stage 1 calls collect_samples() to ping your hosts and write the CSV.
  • Stage 2 calls load_and_aggregate(), compute_statistics(), and create_dashboard() to generate the graph image.
  • Stage 3 calls analyse_dashboard() to send the graph to Claude and print the congestion report.
Add command-line arguments

Python's argparse module lets you pass options when running the script from your terminal. This means you can change the collection duration or target hosts without editing the code.

  • Add the following code below the run_pipeline() function in main.py:
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Network QoS Analyser with Claude AI")
    parser.add_argument("--duration", type=int, default=120, help="Collection duration in seconds (default: 120)")
    parser.add_argument("--interval", type=int, default=2, help="Seconds between ping rounds (default: 2)")
    parser.add_argument("--hosts", type=str, default="8.8.8.8,1.1.1.1,8.8.4.4", help="Comma-separated list of hosts to ping")

    args = parser.parse_args()
    hosts = [h.strip() for h in args.hosts.split(",")]

    run_pipeline(args.duration, args.interval, hosts)

What does this code do?

The if __name__ == "__main__": block only runs when you execute the script directly (not when it is imported by another file).

  • --duration sets how many seconds the collector runs (default 120).
  • --interval sets seconds between each ping round (default 2).
  • --hosts accepts a comma-separated string of IP addresses. The code splits it into a list before passing it to the pipeline.
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

Here is the complete main.py for comparison:

import argparse
import time

from collect_qos import collect_samples
from graph_qos import load_and_aggregate, compute_statistics, create_dashboard
from analyse_with_claude import analyse_dashboard


def run_pipeline(duration, interval, hosts):
    """Run the full QoS analysis pipeline."""
    print("=" * 60)
    print("  NETWORK QoS GRAPHS ANALYSER (with Claude AI)")
    print("=" * 60)

    # Stage 1: Collect data
    print("\n[Stage 1] Collecting real-time QoS data...")
    stage_start = time.time()
    data_file = collect_samples(hosts, duration_seconds=duration, interval_seconds=interval)
    print(f"  Stage 1 completed in {time.time() - stage_start:.1f}s")

    # Stage 2: Generate graphs
    print("\n[Stage 2] Generating QoS dashboard...")
    stage_start = time.time()
    df = load_and_aggregate(data_file)
    stats = compute_statistics(df)
    print("  Statistics:")
    for metric, values in stats.items():
        print(f"    {metric}: mean={values['mean']:.2f}, max={values['max']:.2f}, p95={values['p95']:.2f}")
    dashboard_file = create_dashboard(df)
    print(f"  Stage 2 completed in {time.time() - stage_start:.1f}s")

    # Stage 3: AI analysis
    print("\n[Stage 3] Analysing with Claude AI...")
    stage_start = time.time()
    analysis = analyse_dashboard(dashboard_file, data_file)
    print(f"  Stage 3 completed in {time.time() - stage_start:.1f}s")

    print("\nPipeline complete!")
    return analysis


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Network QoS Analyser with Claude AI")
    parser.add_argument("--duration", type=int, default=120, help="Collection duration in seconds (default: 120)")
    parser.add_argument("--interval", type=int, default=2, help="Seconds between ping rounds (default: 2)")
    parser.add_argument("--hosts", type=str, default="8.8.8.8,1.1.1.1,8.8.4.4", help="Comma-separated list of hosts to ping")

    args = parser.parse_args()
    hosts = [h.strip() for h in args.hosts.split(",")]

    run_pipeline(args.duration, args.interval, hosts)
Run the full pipeline end-to-end

Time to see the whole system in action. You will run the orchestrator and watch it collect fresh data, generate a new dashboard, and get Claude's analysis in one go.

  • Run the pipeline with a shorter duration for a quick test by running this command:
python main.py --duration 60 --interval 2

What does this command do?

This runs the full pipeline with a 60-second collection window instead of the default 120. The --interval 2 keeps the same 2-second gap between ping rounds. You will see Stage 1 collecting data, Stage 2 generating the graph, and Stage 3 printing Claude's congestion analysis.

The default hosts (8.8.8.8, 1.1.1.1, 8.8.4.4) are used automatically when you do not pass --hosts.

The pipeline takes about 60 seconds for data collection, then a few seconds for graphing and API analysis. You should see output showing all three stages completing with timing, followed by Claude's congestion analysis report and token usage.

Why use argparse for a pipeline script?

Command-line arguments let you reuse the same script for different scenarios without editing code. You could monitor a single host for 5 minutes or ping a custom set of internal servers for 10 minutes, all from the same main.py.

For example, to ping only Cloudflare DNS for 30 seconds:

python main.py --duration 30 --hosts 1.1.1.1

Pipeline failing at Stage 3?

If the collector and graph stages succeed but the Claude analysis fails, check that your ANTHROPIC_API_KEY environment variable is still set. Virtual environment activation can sometimes reset variables if you opened a new terminal.

If you see an import error mentioning collect_qos or graph_qos, make sure you are running the command from inside the qos-analyser/ directory where all four .py files live.

Still stuck? Help me debug why my main.py pipeline is failing at a specific stage.

Your entire QoS analysis pipeline is now running end-to-end with a single command. You have a data collector, a graph generator, and an AI analyst working together to monitor your real network. Next, try the Secret Mission to add continuous monitoring with trend detection.

Secret mission

Add Continuous Monitoring with Trend Detection

Your pipeline runs once and gives you a snapshot. But real network monitoring is continuous. What if your tool could watch the network over time, compare windows, and tell you whether things are getting better or worse?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project runs entirely locally. The only ongoing cost is the Anthropic API, which charges per use (not per hour). You will not be billed unless you run the analyser again.

Resources you used:

  • qos-analyser/ project folder on your Desktop (includes source files, virtual environment, collected data CSV, dashboard images, and reports/ directory).
  • ANTHROPIC_API_KEY environment variable (session-only, disappears when you close your terminal).

Keep everything running

No action needed. Choose this if you want to keep experimenting with the QoS analyser or plan to extend it further.

  • Your project files, virtual environment, collected data, and reports all remain in ~/Desktop/qos-analyser/.
  • Your Anthropic account at platform.claude.com remains active. It is pay-per-use, so no charges are incurred unless you run the analyser again.
  • If you closed your terminal, the ANTHROPIC_API_KEY variable is already gone. You will need to re-export it next time you run the pipeline.

Pause - I'll come back to this later

Free up disk space from generated data while keeping your source code intact.

  • Delete the collected data and reports by running:
cd ~/Desktop/qos-analyser
rm -f qos_data.csv qos_dashboard.png
rm -rf reports/
  • Your source files (collect_qos.py, graph_qos.py, analyse_with_claude.py, main.py) and virtual environment remain ready for next time.
  • When you return, activate the virtual environment and re-export your API key before running the pipeline.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and generated data from your machine.

  • Delete the entire project folder by running:

macOS

rm -rf ~/Desktop/qos-analyser

Windows

rmdir /s /q %USERPROFILE%\Desktop\qos-analyser
  • This removes all source files, the virtual environment (including installed packages), collected CSV data, dashboard images, and the reports/ directory.
  • If your terminal is still open, clear the environment variable by running:

macOS

unset ANTHROPIC_API_KEY

Windows

set ANTHROPIC_API_KEY=
  • If you already closed your terminal, the variable is already gone. No further action needed.
  • Your Anthropic account remains active at platform.claude.com. It is pay-per-use with no subscription, so it will not incur charges unless you make API calls.

Nice Work!

Nice Work!

Nice work! You have built a complete AI-powered Network QoS analyser that collects live data from your actual network, visualizes it, and uses Claude to interpret the graphs for congestion patterns.

You've learned how to:

  • Built a real-time QoS data collector that pings hosts on your network and computes latency, jitter, and packet loss from raw ping output.
  • Generated a matplotlib dashboard with threshold lines and anomaly highlighting, saved as a publication-quality PNG.
  • Used Claude's vision API to send graph images for AI-powered congestion analysis and built a full pipeline orchestrator with argparse command-line arguments.
  • Added continuous monitoring with rolling window comparison for trend detection.

Ready to quiz yourself?