| |
| """ |
| Comprehensive Fleet Utilization Benchmark |
| Generates data for research paper Results section. |
| """ |
| import json |
| import sys |
| import os |
| from datetime import datetime |
| from typing import Dict, List, Any, Optional |
| import statistics |
|
|
| |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) |
|
|
| from benchmarks.fleet_utilization.fleet_analyzer import ( |
| FleetUtilizationAnalyzer, |
| FleetUtilizationMetrics, |
| format_metrics_report |
| ) |
|
|
|
|
| class FleetUtilizationBenchmark: |
| """Benchmark fleet utilization across different configurations""" |
| |
| def __init__(self): |
| self.analyzer = FleetUtilizationAnalyzer() |
| self.results = { |
| "metadata": { |
| "generated_at": datetime.now().isoformat(), |
| "system": "Kochi Metro Rail", |
| "analysis_type": "Fleet Utilization" |
| }, |
| "configuration": { |
| "route_length_km": self.analyzer.route_length_km, |
| "avg_speed_kmh": self.analyzer.avg_speed_kmh, |
| "service_hours": self.analyzer.total_service_hours, |
| "peak_headway_minutes": self.analyzer.peak_headway_target, |
| "offpeak_headway_minutes": self.analyzer.offpeak_headway_target |
| }, |
| "fleet_analyses": [], |
| "comparative_analysis": {}, |
| "optimal_fleet": {} |
| } |
| |
| def run_comprehensive_analysis( |
| self, |
| fleet_sizes: Optional[List[int]] = None, |
| maintenance_rate: float = 0.1 |
| ): |
| """ |
| Run comprehensive fleet utilization analysis. |
| |
| Args: |
| fleet_sizes: List of fleet sizes to test (default: 10-40 by 5) |
| maintenance_rate: Percentage of fleet in maintenance |
| """ |
| if fleet_sizes is None: |
| fleet_sizes = [10, 15, 20, 25, 30, 35, 40] |
| |
| print("="*70) |
| print("COMPREHENSIVE FLEET UTILIZATION BENCHMARK") |
| print("="*70) |
| print(f"Fleet Sizes to Test: {fleet_sizes}") |
| print(f"Maintenance Rate: {maintenance_rate*100:.0f}%") |
| print("="*70) |
| print() |
| |
| |
| for i, size in enumerate(fleet_sizes, 1): |
| print(f"[{i}/{len(fleet_sizes)}] Analyzing fleet size: {size} trains...") |
| |
| maintenance_trains = max(1, int(size * maintenance_rate)) |
| metrics = self.analyzer.analyze_fleet_configuration(size, maintenance_trains) |
| |
| |
| result_dict = { |
| "fleet_size": metrics.fleet_size, |
| "minimum_required_trains": metrics.minimum_required_trains, |
| "trains_in_service_peak": metrics.trains_in_service_peak, |
| "trains_in_service_offpeak": metrics.trains_in_service_offpeak, |
| "trains_in_standby": metrics.trains_in_standby, |
| "trains_in_maintenance": metrics.trains_in_maintenance, |
| "peak_demand_coverage_percent": metrics.peak_demand_coverage_percent, |
| "offpeak_demand_coverage_percent": metrics.offpeak_demand_coverage_percent, |
| "overall_coverage_percent": metrics.overall_coverage_percent, |
| "avg_operational_hours_per_train": metrics.avg_operational_hours_per_train, |
| "avg_idle_hours_per_train": metrics.avg_idle_hours_per_train, |
| "utilization_rate_percent": metrics.utilization_rate_percent, |
| "fleet_efficiency_score": metrics.fleet_efficiency_score, |
| "cost_efficiency_score": metrics.cost_efficiency_score |
| } |
| |
| self.results["fleet_analyses"].append(result_dict) |
| |
| print(f" ✓ Coverage: {metrics.overall_coverage_percent:.1f}%") |
| print(f" ✓ Utilization: {metrics.utilization_rate_percent:.1f}%") |
| print(f" ✓ Efficiency: {metrics.fleet_efficiency_score:.1f}/100") |
| print() |
| |
| |
| self._generate_comparative_analysis() |
| |
| |
| self._find_optimal_configuration() |
| |
| print("="*70) |
| print("ANALYSIS COMPLETE") |
| print("="*70) |
| |
| def _generate_comparative_analysis(self): |
| """Generate comparative statistics across all fleet sizes""" |
| analyses = self.results["fleet_analyses"] |
| |
| if not analyses: |
| return |
| |
| |
| coverage = [a["overall_coverage_percent"] for a in analyses] |
| utilization = [a["utilization_rate_percent"] for a in analyses] |
| efficiency = [a["fleet_efficiency_score"] for a in analyses] |
| |
| |
| best_coverage_idx = coverage.index(max(coverage)) |
| best_utilization_idx = utilization.index(max(utilization)) |
| best_efficiency_idx = efficiency.index(max(efficiency)) |
| |
| self.results["comparative_analysis"] = { |
| "coverage_statistics": { |
| "min": min(coverage), |
| "max": max(coverage), |
| "mean": statistics.mean(coverage), |
| "median": statistics.median(coverage), |
| "stdev": statistics.stdev(coverage) if len(coverage) > 1 else 0 |
| }, |
| "utilization_statistics": { |
| "min": min(utilization), |
| "max": max(utilization), |
| "mean": statistics.mean(utilization), |
| "median": statistics.median(utilization), |
| "stdev": statistics.stdev(utilization) if len(utilization) > 1 else 0 |
| }, |
| "efficiency_statistics": { |
| "min": min(efficiency), |
| "max": max(efficiency), |
| "mean": statistics.mean(efficiency), |
| "median": statistics.median(efficiency), |
| "stdev": statistics.stdev(efficiency) if len(efficiency) > 1 else 0 |
| }, |
| "best_performers": { |
| "best_coverage": { |
| "fleet_size": analyses[best_coverage_idx]["fleet_size"], |
| "coverage_percent": analyses[best_coverage_idx]["overall_coverage_percent"] |
| }, |
| "best_utilization": { |
| "fleet_size": analyses[best_utilization_idx]["fleet_size"], |
| "utilization_percent": analyses[best_utilization_idx]["utilization_rate_percent"] |
| }, |
| "best_efficiency": { |
| "fleet_size": analyses[best_efficiency_idx]["fleet_size"], |
| "efficiency_score": analyses[best_efficiency_idx]["fleet_efficiency_score"] |
| } |
| } |
| } |
| |
| def _find_optimal_configuration(self): |
| """Find and store optimal fleet configuration""" |
| print("\nFinding optimal fleet configuration...") |
| |
| optimal_size, optimal_metrics = self.analyzer.find_optimal_fleet_size( |
| min_coverage_required=95.0 |
| ) |
| |
| self.results["optimal_fleet"] = { |
| "optimal_fleet_size": optimal_size, |
| "minimum_required_trains": optimal_metrics.minimum_required_trains, |
| "coverage_percent": optimal_metrics.overall_coverage_percent, |
| "utilization_percent": optimal_metrics.utilization_rate_percent, |
| "efficiency_score": optimal_metrics.fleet_efficiency_score, |
| "cost_efficiency_score": optimal_metrics.cost_efficiency_score, |
| "operational_hours_per_train": optimal_metrics.avg_operational_hours_per_train, |
| "idle_hours_per_train": optimal_metrics.avg_idle_hours_per_train |
| } |
| |
| print(f" ✓ Optimal Fleet Size: {optimal_size} trains") |
| print(f" ✓ Coverage: {optimal_metrics.overall_coverage_percent:.1f}%") |
| print(f" ✓ Efficiency: {optimal_metrics.fleet_efficiency_score:.1f}/100") |
| |
| def save_results(self, filename: Optional[str] = None) -> str: |
| """Save results to JSON file""" |
| if filename is None: |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| filename = f"fleet_utilization_benchmark_{timestamp}.json" |
| |
| filepath = os.path.join( |
| os.path.dirname(os.path.abspath(__file__)), |
| filename |
| ) |
| |
| with open(filepath, 'w') as f: |
| json.dump(self.results, f, indent=2) |
| |
| print(f"\n✓ Results saved to: {filepath}") |
| return filepath |
| |
| def generate_report(self, filename: Optional[str] = None) -> str: |
| """Generate human-readable text report""" |
| if filename is None: |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| filename = f"fleet_utilization_report_{timestamp}.txt" |
| |
| filepath = os.path.join( |
| os.path.dirname(os.path.abspath(__file__)), |
| filename |
| ) |
| |
| with open(filepath, 'w') as f: |
| f.write("="*70 + "\n") |
| f.write("FLEET UTILIZATION BENCHMARK REPORT\n") |
| f.write("="*70 + "\n\n") |
| |
| |
| f.write(f"Generated: {self.results['metadata']['generated_at']}\n") |
| f.write(f"System: {self.results['metadata']['system']}\n\n") |
| |
| |
| f.write("Configuration:\n") |
| f.write("-"*70 + "\n") |
| config = self.results['configuration'] |
| f.write(f" Route Length: {config['route_length_km']} km\n") |
| f.write(f" Average Speed: {config['avg_speed_kmh']} km/h\n") |
| f.write(f" Service Hours: {config['service_hours']} hours/day\n") |
| f.write(f" Peak Headway Target: {config['peak_headway_minutes']} minutes\n") |
| f.write(f" Off-Peak Headway Target: {config['offpeak_headway_minutes']} minutes\n\n") |
| |
| |
| f.write("OPTIMAL FLEET CONFIGURATION:\n") |
| f.write("-"*70 + "\n") |
| optimal = self.results['optimal_fleet'] |
| f.write(f" Optimal Fleet Size: {optimal['optimal_fleet_size']} trains\n") |
| f.write(f" Minimum Required: {optimal['minimum_required_trains']} trains\n") |
| f.write(f" Coverage: {optimal['coverage_percent']:.1f}%\n") |
| f.write(f" Utilization Rate: {optimal['utilization_percent']:.1f}%\n") |
| f.write(f" Fleet Efficiency: {optimal['efficiency_score']:.1f}/100\n") |
| f.write(f" Operational Hours/Train: {optimal['operational_hours_per_train']:.2f} hrs/day\n") |
| f.write(f" Idle Hours/Train: {optimal['idle_hours_per_train']:.2f} hrs/day\n\n") |
| |
| |
| f.write("COMPARATIVE ANALYSIS:\n") |
| f.write("-"*70 + "\n") |
| comp = self.results['comparative_analysis'] |
| |
| f.write("\nCoverage Statistics:\n") |
| stats = comp['coverage_statistics'] |
| f.write(f" Mean: {stats['mean']:.2f}%\n") |
| f.write(f" Median: {stats['median']:.2f}%\n") |
| f.write(f" Range: {stats['min']:.2f}% - {stats['max']:.2f}%\n") |
| f.write(f" Std Dev: {stats['stdev']:.2f}\n") |
| |
| f.write("\nUtilization Statistics:\n") |
| stats = comp['utilization_statistics'] |
| f.write(f" Mean: {stats['mean']:.2f}%\n") |
| f.write(f" Median: {stats['median']:.2f}%\n") |
| f.write(f" Range: {stats['min']:.2f}% - {stats['max']:.2f}%\n") |
| f.write(f" Std Dev: {stats['stdev']:.2f}\n") |
| |
| f.write("\nEfficiency Statistics:\n") |
| stats = comp['efficiency_statistics'] |
| f.write(f" Mean: {stats['mean']:.2f}/100\n") |
| f.write(f" Median: {stats['median']:.2f}/100\n") |
| f.write(f" Range: {stats['min']:.2f} - {stats['max']:.2f}\n") |
| f.write(f" Std Dev: {stats['stdev']:.2f}\n") |
| |
| |
| f.write("\nBest Performers:\n") |
| best = comp['best_performers'] |
| f.write(f" Best Coverage: {best['best_coverage']['fleet_size']} trains ({best['best_coverage']['coverage_percent']:.1f}%)\n") |
| f.write(f" Best Utilization: {best['best_utilization']['fleet_size']} trains ({best['best_utilization']['utilization_percent']:.1f}%)\n") |
| f.write(f" Best Efficiency: {best['best_efficiency']['fleet_size']} trains ({best['best_efficiency']['efficiency_score']:.1f}/100)\n\n") |
| |
| |
| f.write("DETAILED FLEET ANALYSES:\n") |
| f.write("="*70 + "\n\n") |
| |
| for analysis in self.results['fleet_analyses']: |
| f.write(f"Fleet Size: {analysis['fleet_size']} trains\n") |
| f.write("-"*70 + "\n") |
| f.write(f" Minimum Required: {analysis['minimum_required_trains']} trains\n") |
| f.write(f" Peak Service: {analysis['trains_in_service_peak']} trains\n") |
| f.write(f" Off-Peak Service: {analysis['trains_in_service_offpeak']} trains\n") |
| f.write(f" Standby: {analysis['trains_in_standby']} trains\n") |
| f.write(f" Maintenance: {analysis['trains_in_maintenance']} trains\n") |
| f.write(f" Peak Coverage: {analysis['peak_demand_coverage_percent']:.1f}%\n") |
| f.write(f" Off-Peak Coverage: {analysis['offpeak_demand_coverage_percent']:.1f}%\n") |
| f.write(f" Overall Coverage: {analysis['overall_coverage_percent']:.1f}%\n") |
| f.write(f" Operational Hours/Train: {analysis['avg_operational_hours_per_train']:.2f} hrs\n") |
| f.write(f" Idle Hours/Train: {analysis['avg_idle_hours_per_train']:.2f} hrs\n") |
| f.write(f" Utilization Rate: {analysis['utilization_rate_percent']:.1f}%\n") |
| f.write(f" Fleet Efficiency: {analysis['fleet_efficiency_score']:.1f}/100\n") |
| f.write(f" Cost Efficiency: {analysis['cost_efficiency_score']:.1f}/100\n") |
| f.write("\n") |
| |
| print(f"✓ Report saved to: {filepath}") |
| return filepath |
|
|
|
|
| def main(): |
| """Run comprehensive fleet utilization benchmark""" |
| benchmark = FleetUtilizationBenchmark() |
| |
| |
| benchmark.run_comprehensive_analysis( |
| fleet_sizes=[10, 15, 20, 25, 30, 35, 40], |
| maintenance_rate=0.1 |
| ) |
| |
| |
| benchmark.save_results() |
| benchmark.generate_report() |
| |
| print("\n" + "="*70) |
| print("BENCHMARK COMPLETE") |
| print("="*70) |
| print("\nFiles generated:") |
| print(" - fleet_utilization_benchmark_TIMESTAMP.json") |
| print(" - fleet_utilization_report_TIMESTAMP.txt") |
| print("\nUse these results for your research paper Results section!") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
