m2m-python/examples/11_comprehensive_test.py

#!/usr/bin/env python3
import logging
import argparse
import time
from m2m.nbiot import factory

# Configure clean logging output
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%H:%M:%S'
)
logger = logging.getLogger('comprehensive_test')

def run_suite(module, apn):
    """Runs the core network test suite and returns metrics."""
    results = {'rssi': 99, 'dl_kbps': 0.0, 'ul_kbps': 0.0}
    
    print("    - Signal Quality...")
    rssi, ber = module.get_signal_quality()
    results['rssi'] = rssi
    print(f"      RSSI: {rssi}, BER: {ber}")
    
    print("    - DNS Resolution (google.com)...")
    ips = module.dns_query("google.com")
    print(f"      IPs: {ips if ips else 'Failed'}")
    
    if ips:
        print("    - Ping (8.8.8.8)...")
        ping_res = module.ping("8.8.8.8", num=3)
        summary = module.parse_ping_summary(ping_res)
        if summary:
            print(f"      Summary: {summary.get('rcvd')}/{summary.get('sent')} received, RTT avg: {summary.get('avg')}ms")
        else:
            print("      Ping failed.")

        # Symmetrical Speed Tests (150KB each)
        test_size_kb = 150
        
        # Download
        print(f"    - Speed Test (Download {test_size_kb}KB)...")
        module.http_configure(context_id=1, response_header=False)
        if module.http_set_url(f"http://httpbin.org/bytes/{test_size_kb * 1024}"): 
            start = time.time()
            if module.http_get(timeout=300):
                resp = module.http_read_response(wait_time=120)
                duration = time.time() - start
                actual_size_kb = len(resp) / 1024
                if actual_size_kb > 1:
                    speed = (actual_size_kb * 8) / duration
                    results['dl_kbps'] = speed
                    print(f"      DL: {actual_size_kb:.1f} KB in {duration:.1f}s ({speed:.1f} kbps)")
            else: print("      DL test failed.")
        
        # Upload
        print(f"    - Speed Test (Upload {test_size_kb}KB)...")
        dummy_data = b"x" * (test_size_kb * 1024)
        if module.http_set_url("http://httpbin.org/post"):
            start = time.time()
            if module.http_post(dummy_data, timeout=300):
                duration = time.time() - start
                speed = (test_size_kb * 8) / duration
                results['ul_kbps'] = speed
                print(f"      UL: {test_size_kb:.1f} KB in {duration:.1f}s ({speed:.1f} kbps)")
            else: print("      UL test failed.")
            
    return results

def run_comprehensive_test(port, module_type='BG96', apn='iot.telefonica.de'):
    print("\n" + "="*80)
    print(f" MULTI-OPERATOR BENCHMARK: {module_type} on {port}")
    print("="*80)

    module = factory(module_type, port)
    all_results = []

    with module:
        # 1. Basic Health & Identification
        print("\n[1/5] Modem Identification...")
        if not module.check_state():
            print("ERROR: Modem not responding!")
            return
        module.set_echo_mode(True)
        imei = module.read_imei()
        print(f"  IMEI:  {imei}")

        # 2. Operator Search
        print("\n[2/5] Scanning for available operators (this takes a while)...")
        operators = module.search_operators(timeout=180)
        if not operators:
            print("  No operators found or scan failed.")
            return
        
        print(f"  Found {len(operators)} operators.")
        
        # 3. Test Each Operator
        print("\n[3/5] Starting Multi-Operator Test Suite...")
        for i, op in enumerate(operators, 1):
            stat_map = {'0': 'Unknown', '1': 'Available', '2': 'Current', '3': 'Forbidden'}
            op_name = f"{op['long']} ({op['mccmnc']})"
            print(f"\n  ({i}/{len(operators)}) Testing Operator: {op_name} [Status: {stat_map.get(op['status'], op['status'])}]")
            
            if op['status'] == '3':
                print("    Skipping forbidden operator.")
                continue

            if module.set_operator(op['mccmnc'], format=2, act=int(op['act'])):
                print("    - Waiting for registration...")
                start_time = time.time()
                registered = False
                while time.time() - start_time < 90:
                    if module.is_registered():
                        registered = True
                        break
                    time.sleep(2)
                
                if registered:
                    print(f"    - Registered! Attaching and activating context with APN '{apn}'...")
                    if module.attach_network() and module.activate_pdp_context(1):
                        res = run_suite(module, apn)
                        res['name'] = op_name
                        all_results.append(res)
                    else:
                        print("    - Failed to attach or activate PDP.")
                else:
                    print("    - Failed to register on this operator.")
            else:
                print("    - Failed to select operator.")

        # 4. Summary & Ranking
        print("\n" + "="*80)
        print(" FINAL BENCHMARK SUMMARY")
        print("="*80)
        if not all_results:
            print("No network tests were successful.")
        else:
            # Rank by DL Speed + UL Speed
            ranked = sorted(all_results, key=lambda x: (x['dl_kbps'] + x['ul_kbps']), reverse=True)
            
            for i, res in enumerate(ranked, 1):
                star = "★ " if i == 1 else "  "
                print(f"{star}#{i} {res['name']:<25} | DL: {res['dl_kbps']:>6.1f} kbps | UL: {res['ul_kbps']:>6.1f} kbps | RSSI: {res['rssi']}")
            
            best = ranked[0]
            print("\nWINNER: " + best['name'] + " is the best performing operator!")

        # 5. Hardware & Time
        print("\n[5/5] Final Hardware Health...")
        print(f"  Internal Temp: {module.get_temperature()} C")
        bat = module.get_battery_status()
        print(f"  Battery: {bat[1]}%, {bat[2]}mV | Module Clock: {module.get_clock()}")

    print("\n" + "="*80)
    print(" MULTI-OPERATOR BENCHMARK FINISHED")
    print("="*80)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Multi-operator benchmark test.")
    parser.add_argument("port", help="Serial port or socket URL")
    parser.add_argument("--type", default="BG95", help="Module type (BG95, BG96, BC66)")
    parser.add_argument("--apn", default="iot.telefonica.de", help="APN for connection")
    args = parser.parse_args()
    
    run_comprehensive_test(args.port, args.type, args.apn)