Initial commit: Organized structure for future implementations

src/cli.py

@@ -0,0 +1,99 @@
+import csv
+import os
+import sys
+import platform
+import utils.network as network_utils
+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 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(network_utils.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()

src/utils/network.py

@@ -0,0 +1,51 @@
+import socket
+import subprocess
+import platform
+
+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'
+    }