|
|
|
|
|
""" |
|
|
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() |
|
|
|
