Network-Connection-Tester/network_tester.py

import csv
import os
import sys
import platform
import socket
import subprocess
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from rich.console import Console
from rich.table import Table
from rich.live import Live

# Configure the console
console = Console()

def check_ping(hostname):
    """Ping the hostname detecting the current OS."""
    # -n for Windows, -c for Linux/macOS. 1sec / 1000ms Timeout.
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    timeout_param = '-w' if platform.system().lower() == 'windows' else '-W'
    timeout_value = '1000' if platform.system().lower() == 'windows' else '1'
    
    command = ['ping', param, '1', timeout_param, timeout_value, hostname]
    
    try:
        # Return True if the exit code is 0
        return subprocess.call(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0
    except Exception:
        return False

def check_port(hostname, port):
    """Try to open the TCP connection using the specified port."""
    try:
        port = int(port)
    except (ValueError, TypeError):
        return False  # invalid port
        
    try:
        with socket.create_connection((hostname, port), timeout=1.5) as sock:
            return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False

def check_device(device):
    """Execute both test for one device."""
    name = device['name']
    hostname = device['hostname']
    port = device['port']
    
    ping_ok = check_ping(hostname)
    port_ok = check_port(hostname, port) if port else False
    
    return {
        'name': name,
        'hostname': hostname,
        'port': port if port else 'N/A',
        'ping': '[green]ONLINE[/green]' if ping_ok else '[red]OFFLINE[/red]',
        'port_status': '[green]OPEN[/green]' if port_ok else '[red]CLOSED[/red]' if port else 'N/A',
        # Store clean strings for the final CSV export
        'raw_ping': 'ONLINE' if ping_ok else 'OFFLINE',
        'raw_port': 'OPEN' if port_ok else 'CLOSED' if port else 'N/A'
    }

def save_to_csv(results):
    """Export the results to a new CSV file."""
    fieldnames = ['name', 'hostname', 'port', 'ping_status', 'port_status', 'timestamp']
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    name_ts = datetime.now().strftime('%y%m%d%H%M')
    output_filename = "%s_Results.csv" % name_ts
    try:
        with open(output_filename, mode='w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for r in results:
                writer.writerow({
                    'name': r['name'],
                    'hostname': r['hostname'],
                    'port': r['port'],
                    'ping_status': r['raw_ping'],
                    'port_status': r['raw_port'],
                    'timestamp': timestamp
                })
        console.print(f"\n[bold green]✓[/bold green] Results exported to file: [cyan]{output_filename}[/cyan]")
    except Exception as e:
        console.print(f"\n[bold red]Error saving CSV file:[/bold red] {e}")

def main():
    # Subtle header decoration for the brand identity
    if len(sys.argv) > 1:
        input_file = sys.argv[1]
    else:
        input_file = 'devices.csv'
    
    if not os.path.exists(input_file):
        console.print(f"[bold red]Error:[/bold red] Input file could not be found: '{input_file}'")
        return

    # Read target devices from CSV
    devices = []
    with open(input_file, encoding='utf-8', errors='ignore') as f:
        reader = csv.DictReader(f)
        for row in reader:
            devices.append(row)

    results = []

    try:
        while True:
            os.system('cls' if platform.system().lower() == 'windows' else 'clear')  # Clear the console for each iteration
            console.clear()
            # Real-time console interface header
            console.print("\nNetwork Connection Tester - [bold blue]Prime Controls[/bold blue] [dim]v1.0.0[/dim]")
            console.print("[dim]Developed by Emmanuel Hernandez Cruz | Automation Specialist II[/dim]\n")
            console.print(f"\n[bold cyan]●[/bold cyan] Testing {len(devices)} network connections...")
            console.print("Press [bold yellow]Ctrl + C[/bold yellow] to [bold red]stop[/bold red] monitoring and export results.\n")
            # Create dynamic table to show the results in real time
            table = Table(title="Realtime Network Status:")
            table.add_column("Name", justify="left", style="cyan")
            table.add_column("Hostname/IP", justify="left")
            table.add_column("Port", justify="center", style="magenta")
            table.add_column("Ping Status", justify="center")
            table.add_column("Port Status", justify="center")

            results = []  # Reset results for each iteration
    
            # Update the table UI dynamically using Rich Live display
            with Live(table, refresh_per_second=4):
                with ThreadPoolExecutor(max_workers=30) as executor:
                    futures = [executor.submit(check_device, dev) for dev in devices]
                    
                    for future in futures:
                        res = future.result()
                        results.append(res)
                        # Add row to the table
                        table.add_row(res['name'], res['hostname'], str(res['port']), res['ping'], res['port_status'])
            import time
            time.sleep(5)  # Wait for 5 seconds before the next iteration
    except KeyboardInterrupt:
        console.print("\n[bold yellow]<!>[/bold yellow] Monitoring stopped by user.")
        # Generate final report
        save_to_csv(results)
        console.print("\n[dim]Press Enter to close this window...[/dim]", end="")
        input()

if __name__ == '__main__':
    main()