open-signature-generator/benchmark.py

# Written by Jeremy Karst 2025
# This software is licensed under the MIT License with Additional Terms.
# See LICENSE.md for more information.

# Built-in Dependencies
import time
import statistics
from concurrent.futures import ThreadPoolExecutor
import argparse
import random

# External Dependencies
import requests
from rich.console import Console
from rich.table import Table
from rich.progress import Progress


def make_request(base_url, session):
    """Make requests to test the full signature flow"""
    start_time = time.time()
    try:
        # Fetch fonts
        fonts_response = session.get(f"{base_url}/fonts")
        fonts_response.raise_for_status()
        available_fonts = list(fonts_response.json().keys())

        # 1. Register a new signature
        name = f"Test User {random.randint(100000000, 999999999)}"
        register_data = {
            'name': name,
            'font': random.choice(available_fonts),
            'timezone': '0',
            'invert': 'false'
        }
        register_response = session.get(f"{base_url}/register", params=register_data)
        register_response.raise_for_status()
        signature_hash = register_response.text
        register_time = time.time()

        # 2. Verify the signature
        verify_response = session.get(f"{base_url}/verify/{signature_hash}")
        verify_response.raise_for_status()
        verify_time = time.time()

        # 3. Get the signature image
        image_response = session.get(f"{base_url}/images/{signature_hash}.png")
        image_response.raise_for_status()
        image_time = time.time()

        return {
            'register_time': register_time - start_time,
            'register_status': register_response.status_code,
            'verify_time': verify_time - register_time,
            'verify_status': verify_response.status_code,
            'image_time': image_time - verify_time,
            'image_status': image_response.status_code
        }
    except requests.RequestException as e:
        print(f"Request failed: {e}")
        return None

def run_benchmark(base_url, num_threads, requests_per_thread, progress=None):
    """Run benchmark with specified number of threads"""
    session = requests.Session()
    register_times = []
    verify_times = []
    image_times = []
    errors = 0
    
    def worker():
        for _ in range(requests_per_thread):
            result = make_request(base_url, session)
            if result is None:
                nonlocal errors
                errors += 1
            else:
                register_times.append(result['register_time'])
                verify_times.append(result['verify_time'])
                image_times.append(result['image_time'])
            if progress:
                progress.update(task, advance=1)

    # Create and start threads
    with Progress() as progress:
        task = progress.add_task(f"[cyan]Running {num_threads} threads...", total=num_threads * requests_per_thread)
        with ThreadPoolExecutor(max_workers=num_threads) as executor:
            futures = [executor.submit(worker) for _ in range(num_threads)]
            for future in futures:
                future.result()

    if not image_times:
        return None

    return {
        'successful_register_requests': len(register_times),
        'successful_verify_requests': len(verify_times),
        'successful_image_requests': len(image_times),
        'failed_requests': errors,
        'median_register_response_time': statistics.median(register_times),
        'median_verify_response_time': statistics.median(verify_times),
        'median_image_response_time': statistics.median(image_times),
        'min_response_time': min(register_times + verify_times + image_times),
        'max_response_time': max(register_times + verify_times + image_times),
        'total_time': sum(register_times + verify_times + image_times)
    }

def main():
    parser = argparse.ArgumentParser(description='Benchmark web server performance')
    parser.add_argument('--url', default='http://localhost:8080', help='Base URL of the server')
    parser.add_argument('--start-threads', type=int, default=1, help='Starting number of threads')
    parser.add_argument('--max-threads', type=int, default=100, help='Maximum number of threads')
    parser.add_argument('--requests-per-thread', type=int, default=20, help='Number of requests per thread')
    
    args = parser.parse_args()
    
    console = Console()
    results_table = Table(show_header=True, header_style="bold cyan")
    results_table.add_column("Threads")
    results_table.add_column("Total Requests")
    results_table.add_column("Success Rate")
    results_table.add_column("Median Response (s)")
    results_table.add_column("Req/sec")

    console.print(f"\n[bold green]Starting benchmark against {args.url}[/bold green]")
    console.print("[yellow]Note: Each request includes register, verify, and image generation[/yellow]\n")

    # Test thread counts from start to max, doubling every step
    assert args.start_threads > 0, "Starting threads must be greater than 0"
    thread_counts = [args.start_threads]
    while thread_counts[-1] < args.max_threads:
        thread_counts.append(thread_counts[-1] * 2)
    if thread_counts[-1] > args.max_threads:
        thread_counts[-1] = args.max_threads

    for num_threads in thread_counts:
        console.print(f"[yellow]Testing with {num_threads} threads...[/yellow]")
        
        start_time = time.time()
        result = run_benchmark(args.url, num_threads, args.requests_per_thread)
        end_time = time.time()
        
        if result:
            register_success_rate = (result['successful_register_requests'] / (num_threads * args.requests_per_thread)) * 100
            verify_success_rate = (result['successful_verify_requests'] / (num_threads * args.requests_per_thread)) * 100
            image_success_rate = (result['successful_image_requests'] / (num_threads * args.requests_per_thread)) * 100
            requests_per_second = (result['successful_register_requests'] + result['successful_verify_requests'] + result['successful_image_requests']) / (end_time - start_time)
            results_table.add_row(
                str(num_threads),
                str(num_threads * args.requests_per_thread * 3),
                f"{register_success_rate:.1f}% / {verify_success_rate:.1f}% / {image_success_rate:.1f}%",
                f"{result['median_register_response_time']:.3f} / {result['median_verify_response_time']:.3f} / {result['median_image_response_time']:.3f}",
                f"{requests_per_second:.1f}"
            )
        else:
            results_table.add_row(
                str(num_threads),
                str(num_threads * args.requests_per_thread),
                "0%",
                "N/A",
                "N/A"
            )

    console.print("\n[bold]Benchmark Results:[/bold]")
    console.print(results_table)

if __name__ == "__main__":
    main()