Lanbench
def generate_html_report(self) -> str: """Generate HTML report with charts""" template = Template(""" <html> <head><title>LANBench Test Report</title></head> <body> <h1>Network Performance Report</h1> <h2>Summary</h2> <table> <tr><th>Metric</th><th>Value</th></tr> <tr><td>Avg Throughput</td><td>{{ throughput }} Mbps</td></tr> <tr><td>Avg Latency</td><td>{{ latency }} ms</td></tr> <tr><td>Packet Loss</td><td>{{ packet_loss }}%</td></tr> </table> <img src="chart.png" alt="Performance Chart"> </body> </html> """) return template.render(**self.results)
def export_to_csv(self, filename: str): """Export raw data to CSV""" df = pd.DataFrame(self.results['raw_data']) df.to_csv(filename, index=False)
1. Multi-Protocol Testing Suite # protocols.py from enum import Enum import socket import ssl import asyncio class ProtocolType(Enum): TCP = "tcp" UDP = "udp" TLS = "tls" WEBSOCKET = "websocket" QUIC = "quic" LANBench
def collect_system_metrics(self) -> Dict: """Collect real-time system metrics""" return { 'cpu_percent': psutil.cpu_percent(interval=1), 'memory_percent': psutil.virtual_memory().percent, 'network_io': psutil.net_io_counters(), 'connections': len(psutil.net_connections()) }
@app.get("/api/v1/test_status/{test_id}") async def get_test_status(test_id: str): """Get test progress and results""" return get_test_results(test_id) def generate_html_report(self) ->
# Implement throughput measurement pass
def create_latency_chart(self): return go.Figure( data=[go.Scatter(y=list(self.latency_data), mode='lines+markers')], layout=go.Layout(title="Network Latency Over Time") ) # distributed.py import redis import json from typing import List, Dict from multiprocessing import Pool import asyncio class DistributedTester: def init (self, redis_host='localhost', redis_port=6379): self.redis_client = redis.Redis(host=redis_host, port=redis_port) self.test_nodes = [] LANBench Test Report<
@app.websocket("/ws/live") async def websocket_endpoint(websocket: WebSocket): """WebSocket for real-time updates""" await websocket.accept() while True: data = await websocket.receive_text() # Process real-time data await websocket.send_json({"status": "updating", "data": data}) # requirements.txt fastapi==0.104.1 uvicorn==0.24.0 dash==2.14.0 plotly==5.17.0 psutil==5.9.5 numpy==1.24.3 pandas==2.1.0 scapy==2.5.0 redis==5.0.1 websockets==12.0 python-socketio==5.10.0 reportlab==4.0.4 jinja2==3.1.2 Quick Start Example # main.py from fastapi import FastAPI import asyncio async def main(): # Initialize LANBench with all features dashboard = LiveDashboard() api_server = FastAPI() distributed_tester = DistributedTester()
<script> const socket = io('http://localhost:5000'); let chart; function startTest() { const config = { type: document.getElementById('test-type').value, host: document.getElementById('target-host').value, duration: parseInt(document.getElementById('duration').value) }; socket.emit('start_test', config); } socket.on('test_update', (data) => { updateChart(data); updateResults(data); }); </script> </body> </html> # reporting.py import pandas as pd from jinja2 import Template import matplotlib.pyplot as plt from reportlab.lib import colors from reportlab.pdfgen import canvas class ReportGenerator: def init (self, test_results: Dict): self.results = test_results
@app.post("/api/v1/start_test") async def start_test(config: TestConfig, background_tasks: BackgroundTasks): """Start a network test""" test_id = generate_test_id() background_tasks.add_task(run_network_test, test_id, config) return {"test_id": test_id, "status": "started"}
class MetricsCollector: def (self): self.metrics_history: List[NetworkMetrics] = []