WiFi Speeds Tested South America vs Consultant: A Head-to-Head

import iperf3
import json
import time
import logging
from typing import Dict, List, Optional
import argparse

# Configure logging for benchmark traceability
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('wifi_benchmark.log'), logging.StreamHandler()]
)

class WiFiBenchmarker:
    """Runs standardized WiFi speed tests across target access points."""

    def __init__(self, server_host: str, port: int = 5201, duration: int = 10):
        self.server_host = server_host
        self.port = port
        self.duration = duration
        self.results: List[Dict] = []

    def run_throughput_test(self, protocol: str = 'tcp', reverse: bool = False) -> Optional[Dict]:
        """Run iPerf3 throughput test with error handling."""
        try:
            client = iperf3.Client()
            client.server_hostname = self.server_host
            client.port = self.port
            client.duration = self.duration
            client.protocol = protocol
            client.reverse = reverse  # Reverse = download test

            logging.info(f"Starting {protocol.upper()} {'download' if reverse else 'upload'} test to {self.server_host}:{self.port}")
            result = client.run()

            if result.error:
                logging.error(f"iPerf3 test failed: {result.error}")
                return None

            # Extract relevant metrics
            test_result = {
                'timestamp': time.time(),
                'protocol': protocol,
                'direction': 'download' if reverse else 'upload',
                'throughput_mbps': result.sent.Mbps if not reverse else result.received.Mbps,
                'latency_ms': result.latency_ms,
                'packet_loss': result.lost_packets / result.total_packets if result.total_packets > 0 else 0,
                'jitter_ms': result.jitter_ms
            }
            self.results.append(test_result)
            logging.info(f"Test complete: {test_result['throughput_mbps']:.2f} Mbps, {test_result['latency_ms']:.2f} ms latency")
            return test_result

        except Exception as e:
            logging.error(f"Benchmark failed with exception: {str(e)}")
            return None

    def run_full_suite(self, ap_name: str, iterations: int = 5) -> List[Dict]:
        """Run full test suite (TCP up/down, UDP up/down) for a given AP."""
        logging.info(f"Starting full benchmark suite for AP: {ap_name} ({self.server_host})")
        suite_results = []

        for i in range(iterations):
            logging.info(f"Iteration {i+1}/{iterations}")
            # TCP Upload
            tcp_up = self.run_throughput_test(protocol='tcp', reverse=False)
            if tcp_up:
                tcp_up['ap_name'] = ap_name
                suite_results.append(tcp_up)
            time.sleep(2)  # Cooldown between tests

            # TCP Download
            tcp_down = self.run_throughput_test(protocol='tcp', reverse=True)
            if tcp_down:
                tcp_down['ap_name'] = ap_name
                suite_results.append(tcp_down)
            time.sleep(2)

            # UDP Upload
            udp_up = self.run_throughput_test(protocol='udp', reverse=False)
            if udp_up:
                udp_up['ap_name'] = ap_name
                suite_results.append(udp_up)
            time.sleep(2)

            # UDP Download
            udp_down = self.run_throughput_test(protocol='udp', reverse=True)
            if udp_down:
                udp_down['ap_name'] = ap_name
                suite_results.append(udp_down)
            time.sleep(5)

        logging.info(f"Completed suite for {ap_name}: {len(suite_results)} successful tests")
        return suite_results

    def save_results(self, filename: str = 'benchmark_results.json'):
        """Persist results to JSON for later analysis."""
        with open(filename, 'w') as f:
            json.dump(self.results, f, indent=2)
        logging.info(f"Results saved to {filename}")

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='WiFi Benchmark Tool for South America vs Consultant Comparison')
    parser.add_argument('--server', required=True, help='iPerf3 server hostname/IP')
    parser.add_argument('--port', type=int, default=5201, help='iPerf3 server port')
    parser.add_argument('--ap-name', required=True, help='Access Point identifier (e.g., SAM-ISP-SaoPaulo)')
    parser.add_argument('--iterations', type=int, default=5, help='Number of test iterations per direction')
    args = parser.parse_args()

    benchmarker = WiFiBenchmarker(server_host=args.server, port=args.port)
    benchmarker.run_full_suite(ap_name=args.ap_name, iterations=args.iterations)
    benchmarker.save_results()

import json
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from typing import Dict, List
import argparse
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class BenchmarkAnalyzer:
    """Analyzes WiFi benchmark results to compare SAM-ISP vs Consultant deployments."""

    def __init__(self, sam_isp_file: str, consultant_file: str):
        self.sam_isp_df = self._load_and_clean(sam_isp_file, 'SAM-ISP')
        self.consultant_df = self._load_and_clean(consultant_file, 'CD-E')
        self.combined_df = pd.concat([self.sam_isp_df, self.consultant_df], ignore_index=True)

    def _load_and_clean(self, filepath: str, deployment_type: str) -> pd.DataFrame:
        """Load JSON results, clean data, add deployment type column."""
        try:
            with open(filepath, 'r') as f:
                data = json.load(f)
            df = pd.DataFrame(data)
            df['deployment_type'] = deployment_type
            # Convert timestamp to datetime
            df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')
            # Filter out failed tests (throughput < 1 Mbps)
            df = df[df['throughput_mbps'] >= 1.0]
            logging.info(f"Loaded {len(df)} valid tests from {filepath} ({deployment_type})")
            return df
        except FileNotFoundError:
            logging.error(f"File not found: {filepath}")
            raise
        except json.JSONDecodeError:
            logging.error(f"Invalid JSON in {filepath}")
            raise

    def calculate_summary_stats(self) -> pd.DataFrame:
        """Calculate median, p95, p99 stats for throughput, latency, packet loss."""
        stats = self.combined_df.groupby('deployment_type').agg(
            median_throughput=('throughput_mbps', 'median'),
            p95_throughput=('throughput_mbps', lambda x: np.percentile(x, 95)),
            p99_throughput=('throughput_mbps', lambda x: np.percentile(x, 99)),
            median_latency=('latency_ms', 'median'),
            p99_latency=('latency_ms', lambda x: np.percentile(x, 99)),
            median_packet_loss=('packet_loss', 'median'),
            total_tests=('throughput_mbps', 'count')
        ).round(2)
        logging.info(f"Summary stats calculated:\n{stats}")
        return stats

    def plot_throughput_distribution(self, output_file: str = 'throughput_dist.png'):
        """Generate boxplot comparing throughput distributions."""
        plt.figure(figsize=(10, 6))
        sns.boxplot(data=self.combined_df, x='deployment_type', y='throughput_mbps', palette=['#FF6B6B', '#4ECDC4'])
        plt.title('WiFi Throughput Distribution: SAM-ISP vs Consultant Deployments')
        plt.xlabel('Deployment Type')
        plt.ylabel('Throughput (Mbps)')
        plt.tight_layout()
        plt.savefig(output_file, dpi=300)
        logging.info(f"Throughput distribution plot saved to {output_file}")
        plt.close()

    def plot_latency_by_city(self, city_mapping: Dict[str, str], output_file: str = 'latency_by_city.png'):
        """Plot median latency per city for each deployment type."""
        # Add city column from AP name (format: TYPE-CITY-APID)
        self.combined_df['city'] = self.combined_df['ap_name'].apply(
            lambda x: city_mapping.get(x.split('-')[1], 'Unknown') if '-' in x else 'Unknown'
        )
        city_stats = self.combined_df.groupby(['city', 'deployment_type'])['latency_ms'].median().reset_index()

        plt.figure(figsize=(12, 6))
        sns.barplot(data=city_stats, x='city', y='latency_ms', hue='deployment_type', palette=['#FF6B6B', '#4ECDC4'])
        plt.title('Median WiFi Latency by South American City')
        plt.xlabel('City')
        plt.ylabel('Latency (ms)')
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig(output_file, dpi=300)
        logging.info(f"Latency by city plot saved to {output_file}")
        plt.close()

    def calculate_cost_benefit(self, sam_isp_cost_per_mbps: float = 4.20, consultant_cost_per_mbps: float = 18.70) -> Dict:
        """Calculate cost per Mbps and downtime savings."""
        sam_median = self.sam_isp_df['throughput_mbps'].median()
        cons_median = self.consultant_df['throughput_mbps'].median()

        sam_cost_per_mbps = sam_isp_cost_per_mbps
        cons_cost_per_mbps = consultant_cost_per_mbps

        # Assume 10% downtime for SAM-ISP, 0.6% for consultant (from benchmark data)
        sam_downtime_cost = sam_median * 0.10 * 730  # 730 hours per month
        cons_downtime_cost = cons_median * 0.006 * 730

        cost_benefit = {
            'sam_isp_median_throughput': sam_median,
            'consultant_median_throughput': cons_median,
            'sam_isp_monthly_cost_per_mbps': sam_cost_per_mbps,
            'consultant_monthly_cost_per_mbps': cons_cost_per_mbps,
            'sam_isp_downtime_cost_monthly': sam_downtime_cost,
            'consultant_downtime_cost_monthly': cons_downtime_cost,
            'net_monthly_saving': (sam_downtime_cost + (sam_median * sam_cost_per_mbps)) - (cons_downtime_cost + (cons_median * cons_cost_per_mbps))
        }
        logging.info(f"Cost-benefit analysis: {cost_benefit}")
        return cost_benefit

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Analyze WiFi Benchmark Results')
    parser.add_argument('--sam-file', required=True, help='Path to SAM-ISP benchmark JSON')
    parser.add_argument('--consultant-file', required=True, help='Path to Consultant benchmark JSON')
    parser.add_argument('--city-mapping', help='Path to JSON mapping AP ID to city name')
    args = parser.parse_args()

    # Default city mapping for South American capitals
    default_city_mapping = {
        'SaoPaulo': 'São Paulo',
        'BuenosAires': 'Buenos Aires',
        'Santiago': 'Santiago',
        'Lima': 'Lima',
        'Bogota': 'Bogotá',
        'Caracas': 'Caracas'
    }

    analyzer = BenchmarkAnalyzer(args.sam_file, args.consultant_file)
    stats = analyzer.calculate_summary_stats()
    print("Summary Statistics:")
    print(stats)

    analyzer.plot_throughput_distribution()
    analyzer.plot_latency_by_city(city_mapping=default_city_mapping)

    cost_benefit = analyzer.calculate_cost_benefit()
    print("\nCost-Benefit Analysis:")
    for k, v in cost_benefit.items():
        print(f"{k}: {v:.2f}" if isinstance(v, float) else f"{k}: {v}")

import yaml
import subprocess
import re
import logging
from typing import Dict, List, Optional
import argparse

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class WiFiConfigValidator:
    """Validates enterprise WiFi configurations against South America regulatory and performance standards."""

    # South America WiFi 6E allowed channels (5GHz and 6GHz bands)
    ALLOWED_CHANNELS_5GHZ = list(range(36, 165, 4))  # Non-DFS channels for simplicity
    ALLOWED_CHANNELS_6GHZ = list(range(1, 233, 4))   # Full 6GHz spectrum allowed in Brazil/Argentina
    MAX_POWER_5GHZ = 23  # dBm, ANATEL (Brazil) limit
    MAX_POWER_6GHZ = 18  # dBm, ANATEL limit
    REQUIRED_SECURITY = ['WPA3-Personal', 'WPA3-Enterprise']

    def __init__(self, config_path: str):
        self.config_path = config_path
        self.config: Optional[Dict] = None
        self.errors: List[str] = []
        self.warnings: List[str] = []

    def load_config(self) -> bool:
        """Load YAML configuration file for the AP."""
        try:
            with open(self.config_path, 'r') as f:
                self.config = yaml.safe_load(f)
            logging.info(f"Loaded configuration from {self.config_path}")
            return True
        except FileNotFoundError:
            self.errors.append(f"Config file not found: {self.config_path}")
            logging.error(self.errors[-1])
            return False
        except yaml.YAMLError as e:
            self.errors.append(f"Invalid YAML: {str(e)}")
            logging.error(self.errors[-1])
            return False

    def validate_band_settings(self, band: str, band_config: Dict) -> None:
        """Validate 5GHz or 6GHz band configuration."""
        if band not in ['5GHz', '6GHz']:
            self.errors.append(f"Invalid band: {band}")
            return

        # Check channel
        channel = band_config.get('channel')
        if not channel:
            self.errors.append(f"{band}: No channel specified")
        else:
            allowed = self.ALLOWED_CHANNELS_5GHZ if band == '5GHz' else self.ALLOWED_CHANNELS_6GHZ
            if channel not in allowed:
                self.errors.append(f"{band}: Channel {channel} not allowed. Allowed: {allowed[:5]}...")

        # Check transmit power
        power = band_config.get('transmit_power_dbm')
        if not power:
            self.errors.append(f"{band}: No transmit power specified")
        else:
            max_power = self.MAX_POWER_5GHZ if band == '5GHz' else self.MAX_POWER_6GHZ
            if power > max_power:
                self.errors.append(f"{band}: Transmit power {power}dBm exceeds max {max_power}dBm")
            elif power < 10:
                self.warnings.append(f"{band}: Transmit power {power}dBm is below recommended 10dBm")

        # Check channel width
        width = band_config.get('channel_width_mhz')
        if width not in [20, 40, 80, 160]:
            self.errors.append(f"{band}: Invalid channel width {width}MHz. Must be 20/40/80/160")
        if band == '6GHz' and width < 80:
            self.warnings.append(f"6GHz: Channel width {width}MHz is suboptimal, recommend 80/160MHz")

    def validate_security(self) -> None:
        """Validate security settings."""
        security = self.config.get('security', {})
        mode = security.get('mode')
        if not mode:
            self.errors.append("No security mode specified")
        elif mode not in self.REQUIRED_SECURITY:
            self.errors.append(f"Security mode {mode} not allowed. Required: {self.REQUIRED_SECURITY}")

        # Check WPA3 settings
        if mode == 'WPA3-Enterprise':
            radius_server = security.get('radius_server')
            if not radius_server:
                self.errors.append("WPA3-Enterprise: No RADIUS server specified")
            else:
                # Ping RADIUS server to check reachability
                try:
                    subprocess.check_call(['ping', '-c', '1', '-W', '2', radius_server], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                    logging.info(f"RADIUS server {radius_server} is reachable")
                except subprocess.CalledProcessError:
                    self.errors.append(f"WPA3-Enterprise: RADIUS server {radius_server} is unreachable")

    def validate_country_code(self) -> None:
        """Validate country code matches South American deployment."""
        country = self.config.get('country_code')
        valid_countries = ['BR', 'AR', 'CL', 'PE', 'CO', 'VE', 'UY', 'PY', 'BO']
        if not country:
            self.errors.append("No country code specified")
        elif country not in valid_countries:
            self.errors.append(f"Country code {country} not valid for South America deployment. Valid: {valid_countries}")

    def run_all_validations(self) -> bool:
        """Run all validation checks."""
        if not self.load_config():
            return False

        self.validate_country_code()

        # Validate bands
        bands = self.config.get('bands', {})
        for band, band_config in bands.items():
            self.validate_band_settings(band, band_config)

        # Validate security
        self.validate_security()

        # Print results
        if self.errors:
            logging.error("Validation failed with errors:")
            for err in self.errors:
                print(f"ERROR: {err}")
        if self.warnings:
            logging.warning("Validation warnings:")
            for warn in self.warnings:
                print(f"WARNING: {warn}")
        if not self.errors and not self.warnings:
            logging.info("All validations passed!")

        return len(self.errors) == 0

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Validate Enterprise WiFi Configurations for South America')
    parser.add_argument('--config', required=True, help='Path to AP YAML configuration file')
    args = parser.parse_args()

    validator = WiFiConfigValidator(args.config)
    is_valid = validator.run_all_validations()
    exit(0 if is_valid else 1)

client = iperf3.Client()
client.server_hostname = '10.0.0.5'
client.port = 5201
client.duration = 10
client.reverse = True  # Download test
result = client.run()

validator = WiFiConfigValidator('ap_config.yaml')
is_valid = validator.run_all_validations()
if not is_valid:
    raise ValueError("Invalid AP configuration")

df = pd.read_json('benchmark_results.json')
median_throughput = df.groupby('deployment_type')['throughput_mbps'].median()