test_block_optimization.py

#!/usr/bin/env python3
"""Test script to verify block optimization is working properly."""
import json
import sys
sys.path.insert(0, '.')

from DataService.enhanced_generator import EnhancedMetroDataGenerator
from greedyOptim.scheduler import TrainsetSchedulingOptimizer
from greedyOptim.schedule_generator import ScheduleGenerator
from greedyOptim.models import OptimizationConfig

def test_block_optimization():
    """Test that optimizers are actually producing block assignments."""
    
    # Generate test data
    generator = EnhancedMetroDataGenerator()
    data = generator.generate_complete_enhanced_dataset()
    
    # Configure optimizer with block optimization enabled
    config = OptimizationConfig(
        required_service_trains=6,
        min_standby=2,
        optimize_block_assignment=True,
        iterations=5  # Fewer iterations for quick test
    )
    
    print("=" * 60)
    print("TESTING BLOCK OPTIMIZATION")
    print("=" * 60)
    
    optimizer = TrainsetSchedulingOptimizer(data, config)
    
    methods_to_test = ['ga', 'cmaes', 'pso', 'sa', 'nsga2']
    
    results = {}
    for method in methods_to_test:
        print(f"\n{'='*60}")
        print(f"Testing {method.upper()}")
        print("=" * 60)
        
        try:
            result = optimizer.optimize(method=method)
            
            # Check for block assignments
            has_blocks = bool(result.service_block_assignments)
            num_assigned = sum(len(blocks) for blocks in result.service_block_assignments.values()) if has_blocks else 0
            
            print(f"\n{method.upper()} Results:")
            print(f"  - Selected trainsets: {len(result.selected_trainsets)}")
            print(f"  - Has block assignments: {has_blocks}")
            print(f"  - Total blocks assigned: {num_assigned}")
            print(f"  - Fitness score: {result.fitness_score:.2f}")
            
            if has_blocks:
                print(f"  - Block assignments per trainset:")
                for ts_id, blocks in result.service_block_assignments.items():
                    print(f"      {ts_id}: {len(blocks)} blocks")
            
            # Generate schedule using the result
            schedule_gen = ScheduleGenerator(data, config)
            schedule = schedule_gen.generate_schedule(result, method=method, runtime_ms=100)
            
            print(f"\n  Generated Schedule:")
            print(f"    - Schedule ID: {schedule.schedule_id}")
            print(f"    - Trainsets in schedule: {len(schedule.trainsets)}")
            
            # Check service trainsets have blocks
            for trainset in schedule.trainsets:
                if trainset.status.value == "REVENUE_SERVICE":
                    block_count = len(trainset.service_blocks) if trainset.service_blocks else 0
                    total_km = trainset.daily_km_allocation
                    print(f"    - {trainset.trainset_id}: {block_count} blocks, {total_km} km")
            
            results[method] = {
                'success': True,
                'has_blocks': has_blocks,
                'num_blocks': num_assigned,
                'fitness': result.fitness_score
            }
            
        except Exception as e:
            print(f"ERROR with {method}: {e}")
            import traceback
            traceback.print_exc()
            results[method] = {'success': False, 'error': str(e)}
    
    # Summary
    print("\n" + "=" * 60)
    print("SUMMARY")
    print("=" * 60)
    
    all_passed = True
    for method, res in results.items():
        if res['success']:
            status = "✓ PASS" if res['has_blocks'] else "⚠ NO BLOCKS"
            print(f"{method.upper()}: {status} (blocks: {res['num_blocks']}, fitness: {res['fitness']:.2f})")
            if not res['has_blocks']:
                all_passed = False
        else:
            print(f"{method.upper()}: ✗ FAIL ({res['error']})")
            all_passed = False
    
    if all_passed:
        print("\n✓ All optimizers producing block assignments correctly!")
    else:
        print("\n⚠ Some issues detected - check above for details")
    
    return all_passed

if __name__ == "__main__":
    test_block_optimization()