import os
import subprocess
import time
import logging
import smtplib
from email.mime.text import MIMEText
from collections import defaultdict
# Configure logging
logging.basicConfig(filename=’network_performance.log’, level=logging.INFO,
format=’%(asctime)s:%(levelname)s:%(message)s’)
class NetworkMonitor:
def __init__(self, targets, alert_thresholds, check_interval, email_config):
self.targets = targets
self.alert_thresholds = alert_thresholds
self.check_interval = check_interval
self.email_config = email_config
self.results = defaultdict(list)
def ping(self, target):
“””Ping a target to measure latency.”””
try:
output = subprocess.check_output([‘ping’, ‘-c’, ‘4’, target], stderr=subprocess.STDOUT, universal_newlines=True)
latency = self.extract_latency(output)
packet_loss = self.extract_packet_loss(output)
return latency, packet_loss
except subprocess.CalledProcessError as e:
logging.error(f”Ping failed for {target}: {e.output}”)
return None, 100 # Assume 100% packet loss on failure
def extract_latency(self, ping_output):
“””Extract average latency from ping output.”””
lines = ping_output.split(‘\n’)
for line in lines:
if “avg” in line:
avg_latency = float(line.split(‘/’)[4])
return avg_latency
return None
def extract_packet_loss(self, ping_output):
“””Extract packet loss percentage from ping output.”””
lines = ping_output.split(‘\n’)
for line in lines:
if “packet loss” in line:
loss = float(line.split(‘%’)[0].split()[-1])
return loss
return 100 # Assume loss if not found
def measure_bandwidth(self, target):
“””Measure bandwidth using a TCP connection.”””
try:
# Using iperf or similar for bandwidth testing can be implemented here
# For demonstration, we’ll assume a dummy bandwidth result
bandwidth = 100 # Replace with real bandwidth measurement
return bandwidth
except Exception as e:
logging.error(f”Bandwidth measurement failed for {target}: {e}”)
return None
def check_performance(self):
“””Check the performance of each target.”””
for target in self.targets:
latency, packet_loss = self.ping(target)
bandwidth = self.measure_bandwidth(target)
if latency is not None:
self.results[target].append((time.time(), latency, packet_loss, bandwidth))
self.log_and_alert(target, latency, packet_loss, bandwidth)
def log_and_alert(self, target, latency, packet_loss, bandwidth):
“””Log the results and send alerts if thresholds are exceeded.”””
logging.info(f”Target: {target}, Latency: {latency} ms, Packet Loss: {packet_loss}%, Bandwidth: {bandwidth} Mbps”)
if latency > self.alert_thresholds[‘latency’]:
self.send_alert(target, ‘Latency Alert’, f”Latency exceeded: {latency} ms”)
if packet_loss > self.alert_thresholds[‘packet_loss’]:
self.send_alert(target, ‘Packet Loss Alert’, f”Packet loss exceeded: {packet_loss}%”)
# Add bandwidth checks if necessary
def send_alert(self, target, subject, message):
“””Send an email alert.”””
try:
msg = MIMEText(message)
msg[‘Subject’] = subject
msg[‘From’] = self.email_config[‘from’]
msg[‘To’] = self.email_config[‘to’]
with smtplib.SMTP(self.email_config[‘smtp_server’], self.email_config[‘smtp_port’]) as server:
server.starttls()
