I've written many thousands of lines of Python code, and architected complex systems that ingest, transform, normalize, and enrich diverse data for use in AI-driven analytics. But tomorrow morning I need to have a complex command-line utility up and running at 9AM, and at the moment I have zero lines of code.
Time to try vibe coding.
Here's my prompt to Claude Sonnet 4:
You are an expert in Python software development. I need a Python command-line utility that can simultaneously ping multiple IP addresses indefinitely. The list of IP addresses should be read in from a text file which contains one IP address per line. The delay in seconds between pings to each IP address should be controlled by a command line parameter. If that parameter is not passed in, the default delay between pings should be 5 seconds. The execution duration in seconds of this utility should be a command line parameter. If the execution duration is not passed in, then the default duration should be 7200 seconds. Please write the code for me.
And below you will find the 254 lines of code that Claude wrote. I'm going to simplify and edit it, of course. Without Claude, I would be desperately coding through the night and still hammering away at 8:45 AM. With Claude's assistance, I hope to sleep well. :)
"""
Multi-IP Ping Utility
A command-line tool to simultaneously ping multiple IP addresses from a file.
"""
import argparse
import asyncio
import time
import sys
from pathlib import Path
from typing import List, Set
import subprocess
import platform
import ipaddress
class MultiPinger:
def __init__(self, ip_file: str, delay: int = 5, duration: int = 7200):
self.ip_file = ip_file
self.delay = delay
self.duration = duration
self.start_time = time.time()
self.active_ips: Set[str] = set()
def load_ip_addresses(self) -> List[str]:
"""Load and validate IP addresses from file."""
try:
with open(self.ip_file, 'r') as f:
lines = f.readlines()
ip_addresses = []
for line_num, line in enumerate(lines, 1):
ip = line.strip()
if not ip or ip.startswith('#'): # Skip empty lines and comments
continue
try:
# Validate IP address format
ipaddress.ip_address(ip)
ip_addresses.append(ip)
except ipaddress.AddressValueError:
print(f"Warning: Invalid IP address '{ip}' on line {line_num}, skipping...")
if not ip_addresses:
raise ValueError("No valid IP addresses found in file")
return ip_addresses
except FileNotFoundError:
print(f"Error: File '{self.ip_file}' not found")
sys.exit(1)
except Exception as e:
print(f"Error reading file '{self.ip_file}': {e}")
sys.exit(1)
async def ping_ip(self, ip: str) -> None:
"""Continuously ping a single IP address until duration expires."""
self.active_ips.add(ip)
ping_count = 0
# Determine ping command based on OS
if platform.system().lower() == "windows":
ping_cmd = ["ping", "-n", "1", "-w", "3000", ip] # 3 second timeout
else:
ping_cmd = ["ping", "-c", "1", "-W", "3", ip] # 3 second timeout
print(f"Started pinging {ip}")
try:
while time.time() - self.start_time < self.duration:
ping_count += 1
start_ping = time.time()
try:
# Run ping command
result = await asyncio.create_subprocess_exec(
*ping_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await result.communicate()
# Check if ping was successful
if result.returncode == 0:
# Extract response time if possible
output = stdout.decode()
response_time = self.extract_response_time(output)
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
if response_time:
print(f"[{timestamp}] {ip}: Reply received, time={response_time}")
else:
print(f"[{timestamp}] {ip}: Reply received")
else:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {ip}: Request timed out or failed")
except Exception as e:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {ip}: Error during ping - {e}")
# Calculate remaining delay time
elapsed = time.time() - start_ping
remaining_delay = max(0, self.delay - elapsed)
if remaining_delay > 0:
await asyncio.sleep(remaining_delay)
except asyncio.CancelledError:
pass
finally:
self.active_ips.discard(ip)
print(f"Stopped pinging {ip} (sent {ping_count} pings)")
def extract_response_time(self, ping_output: str) -> str:
"""Extract response time from ping output."""
try:
if platform.system().lower() == "windows":
# Windows ping output format
for line in ping_output.split('\n'):
if 'time=' in line.lower():
start = line.lower().find('time=') + 5
end = line.find('ms', start)
if end != -1:
return line[start:end+2]
else:
# Unix/Linux ping output format
for line in ping_output.split('\n'):
if 'time=' in line:
start = line.find('time=') + 5
end = line.find(' ', start)
if end != -1:
return line[start:end]
else:
return line[start:].split()[0]
except:
pass
return ""
async def run(self) -> None:
"""Main execution method."""
print(f"Multi-IP Ping Utility")
print(f"IP file: {self.ip_file}")
print(f"Ping delay: {self.delay} seconds")
print(f"Duration: {self.duration} seconds")
print("-" * 50)
# Load IP addresses
ip_addresses = self.load_ip_addresses()
print(f"Loaded {len(ip_addresses)} IP addresses")
print("-" * 50)
# Create ping tasks for all IPs
tasks = [self.ping_ip(ip) for ip in ip_addresses]
try:
# Run all ping tasks concurrently with timeout
await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=self.duration + 10 # Add buffer for cleanup
)
except asyncio.TimeoutError:
print("\nDuration limit reached, stopping all pings...")
# Cancel all tasks
for task in tasks:
if not task.done():
task.cancel()
# Wait for cancellation to complete
await asyncio.gather(*tasks, return_exceptions=True)
except KeyboardInterrupt:
print("\nInterrupted by user, stopping all pings...")
for task in tasks:
if not task.done():
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
print("-" * 50)
print("Ping utility completed")
def main():
parser = argparse.ArgumentParser(
description="Simultaneously ping multiple IP addresses from a file",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s ips.txt
%(prog)s ips.txt --delay 10
%(prog)s ips.txt --delay 3 --duration 3600
%(prog)s --file servers.txt --delay 15 --duration 1800
"""
)
parser.add_argument(
'file',
nargs='?',
help='Text file containing IP addresses (one per line)'
)
parser.add_argument(
'--file', '-f',
dest='ip_file',
help='Text file containing IP addresses (alternative to positional argument)'
)
parser.add_argument(
'--delay', '-d',
type=int,
default=5,
help='Delay in seconds between pings to each IP (default: 5)'
)
parser.add_argument(
'--duration', '-t',
type=int,
default=7200,
help='Execution duration in seconds (default: 7200)'
)
args = parser.parse_args()
# Determine IP file from positional or named argument
ip_file = args.file or args.ip_file
if not ip_file:
parser.error("IP file must be specified either as positional argument or with --file")
# Validate arguments
if args.delay < 1:
parser.error("Delay must be at least 1 second")
if args.duration < 1:
parser.error("Duration must be at least 1 second")
if not Path(ip_file).exists():
parser.error(f"File '{ip_file}' does not exist")
# Create and run the pinger
pinger = MultiPinger(ip_file, args.delay, args.duration)
try:
asyncio.run(pinger.run())
except KeyboardInterrupt:
print("\nExiting...")
sys.exit(0)
except Exception as e:
print(f"Error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
No comments:
Post a Comment