Network Monitoring PI - Lots of help from ChatGPT. network_metrics.sh could be ran as a cron job, and the network_metrics_display could be a service

network_metrics.sh

#!/bin/bash

# Dependencies: speedtest-cli, mtr, dnsutils (install with: sudo apt install speedtest-cli mtr dnsutils)

# Directory to store JSON files
json_dir="/path/to/json_directory"

# JSON file prefix
json_prefix="network_metrics"

# Rolling JSON file retention (in days)
json_retention_days=1

# Sites to check availability
sites=("google.com" "amazon.com" "hulu.com")

# Current timestamp
timestamp=$(date +"%Y-%m-%d %H:%M:%S")

# Function to check if a command is available
command_exists() {
    command -v "$1" >/dev/null 2>&1
}

# Function to check site availability
check_availability() {
    local site=$1
    local status=""
    if ping -c 5 "$site" >/dev/null; then
        status="Available"
    else
        status="Unavailable"
    fi
    echo "{\"site\":\"$site\",\"status\":\"$status\"}"
}

# Perform availability check for each site
availability_results=""
for site in "${sites[@]}"; do
    availability=$(check_availability "$site")
    availability_results+="$availability,"
done
availability_results=${availability_results::-1}  # Remove trailing comma

# Speedtest
if command_exists "speedtest-cli"; then
    speedtest=$(speedtest-cli --json)
else
    speedtest="{\"error\":\"speedtest-cli command not found\"}"
fi

# Latency
if command_exists "mtr"; then
    latency=$(mtr -c 5 example.com --json)
else
    latency="{\"error\":\"mtr command not found\"}"
fi

# Packet Loss
if command_exists "mtr"; then
    packet_loss=$(mtr -c 100 example.com --json)
else
    packet_loss="{\"error\":\"mtr command not found\"}"
fi

# Jitter (requires 'iperf3' to be installed)
if command_exists "iperf3"; then
    jitter=$(iperf3 -c server_ip -u -b 10M -t 10 -i 1 --json)
else
    jitter="{\"error\":\"iperf3 command not found\"}"
fi

# DNS Resolution Time
if command_exists "dig"; then
    dns_resolution=$(dig +stats +nocomments example.com | awk '/Query time:/ {print $4}' | tr -d '[:alpha:]')
else
    dns_resolution="{\"error\":\"dig command not found\"}"
fi

# Create a JSON object with all the metrics and availability results
metrics_json=$(jq -n \
    --arg timestamp "$timestamp" \
    --argjson speedtest "$speedtest" \
    --argjson latency "$latency" \
    --argjson packet_loss "$packet_loss" \
    --argjson jitter "$jitter" \
    --argjson dns_resolution "$dns_resolution" \
    --argjson availability_results "[$availability_results]" \
    '{timestamp: $timestamp, speedtest: $speedtest, latency: $latency, packet_loss: $packet_loss, jitter: $jitter, dns_resolution: $dns_resolution, availability: $availability_results}')

# Output the values passed to jq
echo "Timestamp: $timestamp"
echo "Speedtest: $speedtest"
echo "Latency: $latency"
echo "Packet Loss: $packet_loss"
echo "Jitter: $jitter"
echo "DNS Resolution: $dns_resolution"
echo "Availability Results: $availability

network_metrics_display.py

import json
import os
from RPLCD.i2c import CharLCD
import time

# LCD constants
LCD_ADDRESS = 0x27  # I2C address of the LCD
LCD_WIDTH = 16
LCD_ROWS = 2

# Load data from JSON file
def load_json_data(json_dir):
    with open(json_dir, "r") as json_file:
        data = json.load(json_file)
    return data

# Format download and upload speeds to Mbps
def format_speed(speed_bps):
    speed_mbps = speed_bps / 1_000_000
    speed_mbps_str = f"{speed_mbps:.0f} Mbps"
    if len(speed_mbps_str) > LCD_WIDTH:
        speed_mbps_str = speed_mbps_str[:-5]  # Truncate " Mbps"
    return speed_mbps_str

# Format jitter value with 4 decimal places
def format_jitter(jitter_ms):
    return f"{jitter_ms:.4f} ms"

# Main program
if __name__ == "__main__":
    # Initialize LCD
    lcd = CharLCD(i2c_expander='PCF8574', address=LCD_ADDRESS, port=1, cols=LCD_WIDTH, rows=LCD_ROWS)
    lcd.clear()

    # Construct JSON file path
    json_dir = "/path/to/network_metrics/current.json"

    # Sleep time between strings
    sleep_time = 5  # Set the desired sleep time (in seconds)

    while True:
        # Load JSON data
        data = load_json_data(json_dir)

        # Parse desired values from JSON
        download = format_speed(data["speedtest"]["download"])
        upload = format_speed(data["speedtest"]["upload"])
        jitter = format_jitter(data["jitter"]["end"]["streams"][0]["udp"]["jitter_ms"])
        dns_query = str(data["dns_resolution"]["query_time"]) + " ms"
        google_status = data["availability"][0]["status"]
        amazon_status = data["availability"][1]["status"]
        hulu_status = data["availability"][2]["status"]
        ip_address = data.get("ip_address", "N/A")  # Use default value if "ip_address" is missing
        ssid = data.get("ssid")
        if ssid is None or ssid.strip() == "":
            ssid = "N/A"
        public_ip = data.get("public_ip", "N/A")  # Use default value if "public_ip" is missing

        # Display data on LCD
        lcd.clear()
        lcd.home()
        lcd.write_string("Public IP:\n\r" + public_ip)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("Download:\n\r" + download)
        time.sleep(sleep_time)


        lcd.clear()
        lcd.home()
        lcd.write_string("Upload:\n\r" + upload)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("Jitter:\n\r" + jitter)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("DNS Query:\n\r" + dns_query)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("Google:\n\r" + google_status)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("Amazon:\n\r" + amazon_status)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("Hulu:\n\r" + hulu_status)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("IP Addr:\n\r" + ip_address)
        time.sleep(sleep_time)

        lcd.clear()
        lcd.home()
        lcd.write_string("SSID:\n\r" + ssid)
        time.sleep(sleep_time)