proper scheduling-1, better iterations set

greedyOptim/advanced_optimizers.py

@@ -3,7 +3,7 @@ Advanced optimization algorithms for trainset scheduling.
 Includes CMA-ES, Particle Swarm Optimization, and Simulated Annealing.
 """
 import numpy as np
-from typing import Optional
+from typing import Optional, Dict, List
 import math
 
 from .models import OptimizationResult, OptimizationConfig
@@ -17,6 +17,8 @@ class CMAESOptimizer:
         self.evaluator = evaluator
         self.config = config or OptimizationConfig()
         self.n = evaluator.num_trainsets
+        self.n_blocks = evaluator.num_blocks
+        self.optimize_blocks = self.config.optimize_block_assignment
         self.lam = self.config.population_size  # Population size
         self.mu = self.config.population_size // 2  # Number of parents
         
@@ -47,12 +49,41 @@ class CMAESOptimizer:
         """Decode continuous values to discrete actions."""
         return np.clip(np.round(x), 0, 2).astype(int)
     
-    def optimize(self, generations: int = 150) -> OptimizationResult:
+    def _create_block_assignment(self, trainset_sol: np.ndarray) -> np.ndarray:
+        """Create optimized block assignments for a trainset solution."""
+        service_indices = np.where(trainset_sol == 0)[0]
+        
+        if len(service_indices) == 0:
+            return np.full(self.n_blocks, -1, dtype=int)
+        
+        # Distribute blocks evenly with some randomization
+        block_sol = np.zeros(self.n_blocks, dtype=int)
+        for i in range(self.n_blocks):
+            block_sol[i] = service_indices[i % len(service_indices)]
+        
+        # Random shuffle to explore different assignments
+        np.random.shuffle(block_sol)
+        
+        # Repair to ensure valid assignments
+        for i in range(len(block_sol)):
+            if block_sol[i] not in service_indices:
+                block_sol[i] = np.random.choice(service_indices)
+        
+        return block_sol
+    
+    def optimize(self, generations: Optional[int] = None) -> OptimizationResult:
         """Run CMA-ES optimization."""
+        # Use config.iterations as default if not specified
+        if generations is None:
+            generations = self.config.iterations * 15  # Scale iterations for CMA-ES
+        
         best_fitness = float('inf')
         best_solution: Optional[np.ndarray] = None
+        best_block_solution: Optional[np.ndarray] = None
         
         print(f"Starting CMA-ES optimization for {generations} generations")
+        if self.optimize_blocks:
+            print(f"Optimizing block assignments for {self.n_blocks} service blocks")
         
         for gen in range(generations):
             try:
@@ -68,10 +99,20 @@ class CMAESOptimizer:
                 # Evaluate
                 fitness = []
                 decoded_pop = []
+                block_pop = []
+                
                 for ind in population:
                     decoded = self._decode(ind)
                     decoded_pop.append(decoded)
-                    fitness.append(self.evaluator.fitness_function(decoded))
+                    
+                    if self.optimize_blocks:
+                        block_sol = self._create_block_assignment(decoded)
+                        block_pop.append(block_sol)
+                        fit = self.evaluator.schedule_fitness_function(decoded, block_sol)
+                    else:
+                        fit = self.evaluator.fitness_function(decoded)
+                    
+                    fitness.append(fit)
                 
                 fitness = np.array(fitness)
                 decoded_pop = np.array(decoded_pop)
@@ -81,6 +122,8 @@ class CMAESOptimizer:
                 if fitness[gen_best_idx] < best_fitness:
                     best_fitness = fitness[gen_best_idx]
                     best_solution = decoded_pop[gen_best_idx].copy()
+                    if self.optimize_blocks and block_pop:
+                        best_block_solution = block_pop[gen_best_idx].copy()
                 
                 if gen % 30 == 0:
                     print(f"Generation {gen}: Best Fitness = {best_fitness:.2f}")
@@ -117,9 +160,10 @@ class CMAESOptimizer:
         if best_solution is None:
             raise RuntimeError("No valid solution found during CMA-ES optimization")
         
-        return self._build_result(best_solution, best_fitness)
+        return self._build_result(best_solution, best_fitness, best_block_solution)
     
-    def _build_result(self, solution: np.ndarray, fitness: float) -> OptimizationResult:
+    def _build_result(self, solution: np.ndarray, fitness: float,
+                      block_solution: Optional[np.ndarray] = None) -> OptimizationResult:
         """Build optimization result from solution."""
         objectives = self.evaluator.calculate_objectives(solution)
         
@@ -132,13 +176,27 @@ class CMAESOptimizer:
             valid, reason = self.evaluator.check_hard_constraints(ts_id)
             explanations[ts_id] = "✓ Fit for service" if valid else f"⚠ {reason}"
         
+        # Build block assignments
+        block_assignments = {}
+        if block_solution is not None and self.optimize_blocks:
+            for ts_id in service:
+                block_assignments[ts_id] = []
+            
+            for block_idx, train_idx in enumerate(block_solution):
+                if 0 <= train_idx < len(self.evaluator.trainsets):
+                    ts_id = self.evaluator.trainsets[train_idx]
+                    if ts_id in block_assignments:
+                        block_id = self.evaluator.all_blocks[block_idx]['block_id']
+                        block_assignments[ts_id].append(block_id)
+        
         return OptimizationResult(
             selected_trainsets=service,
             standby_trainsets=standby,
             maintenance_trainsets=maintenance,
             objectives=objectives,
             fitness_score=fitness,
-            explanation=explanations
+            explanation=explanations,
+            service_block_assignments=block_assignments
         )
 
 
@@ -150,6 +208,8 @@ class ParticleSwarmOptimizer:
         self.config = config or OptimizationConfig()
         self.n_particles = self.config.population_size
         self.n_dimensions = evaluator.num_trainsets
+        self.n_blocks = evaluator.num_blocks
+        self.optimize_blocks = self.config.optimize_block_assignment
         
         # PSO parameters
         self.w = 0.7  # Inertia weight
@@ -161,8 +221,25 @@ class ParticleSwarmOptimizer:
         """Decode continuous values to discrete actions."""
         return np.clip(np.round(x), 0, 2).astype(int)
     
-    def optimize(self, generations: int = 200) -> OptimizationResult:
+    def _create_block_assignment(self, trainset_sol: np.ndarray) -> np.ndarray:
+        """Create block assignments for a trainset solution."""
+        service_indices = np.where(trainset_sol == 0)[0]
+        
+        if len(service_indices) == 0:
+            return np.full(self.n_blocks, -1, dtype=int)
+        
+        block_sol = np.zeros(self.n_blocks, dtype=int)
+        for i in range(self.n_blocks):
+            block_sol[i] = service_indices[i % len(service_indices)]
+        
+        return block_sol
+    
+    def optimize(self, generations: Optional[int] = None) -> OptimizationResult:
         """Run PSO optimization."""
+        # Use config.iterations as default if not specified
+        if generations is None:
+            generations = self.config.iterations * 20  # Scale iterations for PSO
+        
         # Initialize particles
         positions = np.random.uniform(0, 3, (self.n_particles, self.n_dimensions))
         velocities = np.random.uniform(-1, 1, (self.n_particles, self.n_dimensions))
@@ -170,29 +247,41 @@ class ParticleSwarmOptimizer:
         # Personal best positions and fitness
         p_best_positions = positions.copy()
         p_best_fitness = np.array([float('inf')] * self.n_particles)
+        p_best_blocks = [None] * self.n_particles
         
         # Global best
         g_best_position = np.zeros(self.n_dimensions)
         g_best_fitness = float('inf')
+        g_best_block = None
         
         print(f"Starting PSO optimization with {self.n_particles} particles for {generations} generations")
+        if self.optimize_blocks:
+            print(f"Optimizing block assignments for {self.n_blocks} service blocks")
         
         for gen in range(generations):
             try:
                 for i in range(self.n_particles):
                     # Evaluate particle
                     decoded = self._decode(positions[i])
-                    fitness = self.evaluator.fitness_function(decoded)
+                    
+                    if self.optimize_blocks:
+                        block_sol = self._create_block_assignment(decoded)
+                        fitness = self.evaluator.schedule_fitness_function(decoded, block_sol)
+                    else:
+                        block_sol = None
+                        fitness = self.evaluator.fitness_function(decoded)
                     
                     # Update personal best
                     if fitness < p_best_fitness[i]:
                         p_best_fitness[i] = fitness
                         p_best_positions[i] = positions[i].copy()
+                        p_best_blocks[i] = block_sol.copy() if block_sol is not None else None
                         
                         # Update global best
                         if fitness < g_best_fitness:
                             g_best_fitness = fitness
                             g_best_position = positions[i].copy()
+                            g_best_block = block_sol.copy() if block_sol is not None else None
                 
                 # Update velocities and positions
                 for i in range(self.n_particles):
@@ -217,9 +306,10 @@ class ParticleSwarmOptimizer:
                 break
         
         best_solution = self._decode(g_best_position)
-        return self._build_result(best_solution, g_best_fitness)
+        return self._build_result(best_solution, g_best_fitness, g_best_block)
     
-    def _build_result(self, solution: np.ndarray, fitness: float) -> OptimizationResult:
+    def _build_result(self, solution: np.ndarray, fitness: float,
+                      block_solution: Optional[np.ndarray] = None) -> OptimizationResult:
         """Build optimization result from solution."""
         objectives = self.evaluator.calculate_objectives(solution)
         
@@ -232,13 +322,27 @@ class ParticleSwarmOptimizer:
             valid, reason = self.evaluator.check_hard_constraints(ts_id)
             explanations[ts_id] = "✓ Fit for service" if valid else f"⚠ {reason}"
         
+        # Build block assignments
+        block_assignments = {}
+        if block_solution is not None and self.optimize_blocks:
+            for ts_id in service:
+                block_assignments[ts_id] = []
+            
+            for block_idx, train_idx in enumerate(block_solution):
+                if 0 <= train_idx < len(self.evaluator.trainsets):
+                    ts_id = self.evaluator.trainsets[train_idx]
+                    if ts_id in block_assignments:
+                        block_id = self.evaluator.all_blocks[block_idx]['block_id']
+                        block_assignments[ts_id].append(block_id)
+        
         return OptimizationResult(
             selected_trainsets=service,
             standby_trainsets=standby,
             maintenance_trainsets=maintenance,
             objectives=objectives,
             fitness_score=fitness,
-            explanation=explanations
+            explanation=explanations,
+            service_block_assignments=block_assignments
         )
 
 
@@ -249,6 +353,8 @@ class SimulatedAnnealingOptimizer:
         self.evaluator = evaluator
         self.config = config or OptimizationConfig()
         self.n_dimensions = evaluator.num_trainsets
+        self.n_blocks = evaluator.num_blocks
+        self.optimize_blocks = self.config.optimize_block_assignment
         
     def _get_neighbor(self, solution: np.ndarray) -> np.ndarray:
         """Generate a neighbor solution by randomly changing one gene."""
@@ -257,26 +363,80 @@ class SimulatedAnnealingOptimizer:
         neighbor[idx] = np.random.randint(0, 3)
         return neighbor
     
+    def _get_block_neighbor(self, block_sol: np.ndarray, service_indices: np.ndarray) -> np.ndarray:
+        """Generate a neighbor block assignment."""
+        neighbor = block_sol.copy()
+        
+        if len(service_indices) == 0:
+            return neighbor
+        
+        # Randomly reassign a few blocks
+        num_changes = max(1, self.n_blocks // 20)
+        for _ in range(num_changes):
+            idx = np.random.randint(0, len(neighbor))
+            neighbor[idx] = np.random.choice(service_indices)
+        
+        return neighbor
+    
+    def _create_block_assignment(self, trainset_sol: np.ndarray) -> np.ndarray:
+        """Create block assignments for a trainset solution."""
+        service_indices = np.where(trainset_sol == 0)[0]
+        
+        if len(service_indices) == 0:
+            return np.full(self.n_blocks, -1, dtype=int)
+        
+        block_sol = np.zeros(self.n_blocks, dtype=int)
+        for i in range(self.n_blocks):
+            block_sol[i] = service_indices[i % len(service_indices)]
+        
+        return block_sol
+    
     def _temperature(self, iteration: int, max_iterations: int) -> float:
         """Calculate temperature using exponential cooling."""
         return 100.0 * (0.95 ** iteration)
     
-    def optimize(self, max_iterations: int = 10000) -> OptimizationResult:
+    def optimize(self, max_iterations: Optional[int] = None) -> OptimizationResult:
         """Run Simulated Annealing optimization."""
+        # Use config.iterations as default if not specified
+        if max_iterations is None:
+            max_iterations = self.config.iterations * 1000  # Scale iterations for SA
+        
         # Initialize with a random solution
         current_solution = np.random.randint(0, 3, self.n_dimensions)
-        current_fitness = self.evaluator.fitness_function(current_solution)
+        current_block_sol = self._create_block_assignment(current_solution) if self.optimize_blocks else None
+        
+        if self.optimize_blocks and current_block_sol is not None:
+            current_fitness = self.evaluator.schedule_fitness_function(current_solution, current_block_sol)
+        else:
+            current_fitness = self.evaluator.fitness_function(current_solution)
         
         best_solution = current_solution.copy()
+        best_block_sol = current_block_sol.copy() if current_block_sol is not None else None
         best_fitness = current_fitness
         
         print(f"Starting Simulated Annealing optimization for {max_iterations} iterations")
+        if self.optimize_blocks:
+            print(f"Optimizing block assignments for {self.n_blocks} service blocks")
         
         for iteration in range(max_iterations):
             try:
                 # Generate neighbor
                 neighbor = self._get_neighbor(current_solution)
-                neighbor_fitness = self.evaluator.fitness_function(neighbor)
+                
+                if self.optimize_blocks:
+                    service_indices = np.where(neighbor == 0)[0]
+                    # Sometimes create new block assignments, sometimes just modify
+                    if np.random.random() < 0.3:
+                        neighbor_block_sol = self._create_block_assignment(neighbor)
+                    else:
+                        neighbor_block_sol = self._get_block_neighbor(
+                            current_block_sol if current_block_sol is not None else self._create_block_assignment(neighbor),
+                            service_indices
+                        )
+                    neighbor_fitness = self.evaluator.schedule_fitness_function(neighbor, neighbor_block_sol)
+                else:
+                    neighbor_block_sol = None
+                    neighbor_fitness = self.evaluator.fitness_function(neighbor)
                 
                 # Calculate acceptance probability
                 temperature = self._temperature(iteration, max_iterations)
@@ -285,6 +445,7 @@ class SimulatedAnnealingOptimizer:
                     # Accept better solution
                     current_solution = neighbor
                     current_fitness = neighbor_fitness
+                    current_block_sol = neighbor_block_sol
                 elif temperature > 0:
                     # Accept worse solution with probability
                     delta = neighbor_fitness - current_fitness
@@ -292,11 +453,13 @@ class SimulatedAnnealingOptimizer:
                     if np.random.random() < probability:
                         current_solution = neighbor
                         current_fitness = neighbor_fitness
+                        current_block_sol = neighbor_block_sol
                 
                 # Update best solution
                 if current_fitness < best_fitness:
                     best_solution = current_solution.copy()
                     best_fitness = current_fitness
+                    best_block_sol = current_block_sol.copy() if current_block_sol is not None else None
                 
                 if iteration % 1000 == 0:
                     print(f"Iteration {iteration}: Best Fitness = {best_fitness:.2f}, Temperature = {temperature:.2f}")
@@ -305,9 +468,10 @@ class SimulatedAnnealingOptimizer:
                 print(f"Error in SA iteration {iteration}: {e}")
                 break
         
-        return self._build_result(best_solution, best_fitness)
+        return self._build_result(best_solution, best_fitness, best_block_sol)
     
-    def _build_result(self, solution: np.ndarray, fitness: float) -> OptimizationResult:
+    def _build_result(self, solution: np.ndarray, fitness: float,
+                      block_solution: Optional[np.ndarray] = None) -> OptimizationResult:
         """Build optimization result from solution."""
         objectives = self.evaluator.calculate_objectives(solution)
         
@@ -320,11 +484,25 @@ class SimulatedAnnealingOptimizer:
             valid, reason = self.evaluator.check_hard_constraints(ts_id)
             explanations[ts_id] = "✓ Fit for service" if valid else f"⚠ {reason}"
         
+        # Build block assignments
+        block_assignments = {}
+        if block_solution is not None and self.optimize_blocks:
+            for ts_id in service:
+                block_assignments[ts_id] = []
+            
+            for block_idx, train_idx in enumerate(block_solution):
+                if 0 <= train_idx < len(self.evaluator.trainsets):
+                    ts_id = self.evaluator.trainsets[train_idx]
+                    if ts_id in block_assignments:
+                        block_id = self.evaluator.all_blocks[block_idx]['block_id']
+                        block_assignments[ts_id].append(block_id)
+        
         return OptimizationResult(
             selected_trainsets=service,
             standby_trainsets=standby,
             maintenance_trainsets=maintenance,
             objectives=objectives,
             fitness_score=fitness,
-            explanation=explanations
+            explanation=explanations,
+            service_block_assignments=block_assignments
         )

greedyOptim/evaluator.py

@@ -7,6 +7,7 @@ from datetime import datetime
 from typing import Dict, List, Tuple, Optional
 
 from .models import OptimizationConfig, TrainsetConstraints
+from .service_blocks import ServiceBlockGenerator
 
 
 # Status normalization mappings (backend format -> internal format)
@@ -50,6 +51,11 @@ class TrainsetSchedulingEvaluator:
         self.trainsets = [ts['trainset_id'] for ts in data['trainset_status']]
         self.num_trainsets = len(self.trainsets)
         
+        # Service block generator for schedule optimization
+        self.block_generator = ServiceBlockGenerator()
+        self.all_blocks = self.block_generator.get_all_service_blocks()
+        self.num_blocks = len(self.all_blocks)
+        
         # Build lookup dictionaries
         self._build_lookups()
         
@@ -276,4 +282,146 @@ class TrainsetSchedulingEvaluator:
             obj['constraint_penalty'] * 5.0            # Minimize (positive weight)
         )
         
-        return fitness
+        return fitness
+    
+    def evaluate_schedule_quality(self, service_trains: List[str], 
+                                   block_assignments: Dict[str, List[int]]) -> Dict[str, float]:
+        """Evaluate schedule quality objectives.
+        
+        Args:
+            service_trains: List of trainset IDs in service
+            block_assignments: Maps trainset_id -> list of block indices
+            
+        Returns:
+            Dictionary with schedule quality scores
+        """
+        scores = {
+            'headway_consistency': 0.0,
+            'service_coverage': 0.0,
+            'block_distribution': 0.0,
+            'peak_coverage': 0.0
+        }
+        
+        if not block_assignments:
+            return scores
+        
+        # Flatten all assigned block indices
+        all_assigned_blocks = set()
+        blocks_per_train = []
+        
+        for ts_id, block_indices in block_assignments.items():
+            all_assigned_blocks.update(block_indices)
+            blocks_per_train.append(len(block_indices))
+        
+        # 1. Service Coverage: What % of blocks are covered?
+        coverage = len(all_assigned_blocks) / self.num_blocks if self.num_blocks > 0 else 0
+        scores['service_coverage'] = coverage * 100.0
+        
+        # 2. Peak Coverage: Are peak blocks covered?
+        peak_indices = self.block_generator.get_peak_block_indices()
+        covered_peak = len(all_assigned_blocks.intersection(peak_indices))
+        peak_coverage = covered_peak / len(peak_indices) if peak_indices else 0
+        scores['peak_coverage'] = peak_coverage * 100.0
+        
+        # 3. Block Distribution: Are blocks evenly distributed across trains?
+        if blocks_per_train and len(blocks_per_train) > 1:
+            std_dev = float(np.std(blocks_per_train))
+            mean_blocks = float(np.mean(blocks_per_train))
+            cv = std_dev / mean_blocks if mean_blocks > 0 else 1.0
+            # Lower CV = better distribution (100 - penalty)
+            scores['block_distribution'] = max(0, 100.0 - cv * 50.0)
+        else:
+            scores['block_distribution'] = 100.0
+        
+        # 4. Headway Consistency: Check departure gaps
+        scores['headway_consistency'] = self._evaluate_headway_consistency(all_assigned_blocks)
+        
+        return scores
+    
+    def _evaluate_headway_consistency(self, assigned_block_indices: set) -> float:
+        """Evaluate headway consistency for assigned blocks.
+        
+        Args:
+            assigned_block_indices: Set of block indices that are covered
+            
+        Returns:
+            Headway consistency score (0-100)
+        """
+        if not assigned_block_indices:
+            return 0.0
+        
+        # Get departure times of assigned blocks
+        departure_minutes = []
+        for idx in assigned_block_indices:
+            if idx < len(self.all_blocks):
+                block = self.all_blocks[idx]
+                time_str = block['departure_time']
+                hour, minute = map(int, time_str.split(':'))
+                departure_minutes.append(hour * 60 + minute)
+        
+        if len(departure_minutes) < 2:
+            return 50.0  # Not enough data
+        
+        # Sort and calculate gaps
+        departure_minutes.sort()
+        gaps = []
+        for i in range(1, len(departure_minutes)):
+            gaps.append(departure_minutes[i] - departure_minutes[i-1])
+        
+        if not gaps:
+            return 50.0
+        
+        # Calculate coefficient of variation for gaps
+        mean_gap = float(np.mean(gaps))
+        std_gap = float(np.std(gaps))
+        
+        # Lower CV = more consistent headways
+        cv = std_gap / mean_gap if mean_gap > 0 else 1.0
+        
+        # Score: 100 for perfect consistency (CV=0), decreasing with higher CV
+        score = max(0, 100.0 - cv * 100.0)
+        
+        return score
+    
+    def schedule_fitness_function(self, trainset_solution: np.ndarray, 
+                                   block_solution: np.ndarray) -> float:
+        """Combined fitness function for trainset and block assignment optimization.
+        
+        Args:
+            trainset_solution: Array where trainset_solution[i] = 0/1/2 (service/standby/maint)
+            block_solution: Array where block_solution[j] = trainset_index or -1 (unassigned)
+            
+        Returns:
+            Combined fitness score (lower is better)
+        """
+        # First evaluate trainset selection
+        base_fitness = self.fitness_function(trainset_solution)
+        
+        # Decode service trains
+        service_train_indices = [i for i, v in enumerate(trainset_solution) if v == 0]
+        service_trains = [self.trainsets[i] for i in service_train_indices]
+        
+        # Build block assignments
+        block_assignments = {}
+        for ts_idx in service_train_indices:
+            ts_id = self.trainsets[ts_idx]
+            block_assignments[ts_id] = []
+        
+        for block_idx, assigned_train_idx in enumerate(block_solution):
+            if assigned_train_idx >= 0 and assigned_train_idx < len(self.trainsets):
+                ts_id = self.trainsets[int(assigned_train_idx)]
+                if ts_id in block_assignments:
+                    block_assignments[ts_id].append(block_idx)
+        
+        # Evaluate schedule quality
+        schedule_scores = self.evaluate_schedule_quality(service_trains, block_assignments)
+        
+        # Add schedule objectives to fitness
+        schedule_penalty = (
+            -(schedule_scores['service_coverage'] * 1.5) +     # Maximize coverage
+            -(schedule_scores['peak_coverage'] * 2.0) +        # Maximize peak coverage
+            -(schedule_scores['block_distribution'] * 1.0) +   # Maximize even distribution
+            -(schedule_scores['headway_consistency'] * 1.0)    # Maximize consistency
+        )
+        
+        return base_fitness + schedule_penalty

greedyOptim/genetic_algorithm.py

@@ -2,7 +2,7 @@
 Genetic Algorithm optimizer for trainset scheduling.
 """
 import numpy as np
-from typing import Tuple, Optional
+from typing import Tuple, Optional, Dict, List
 
 from .models import OptimizationResult, OptimizationConfig
 from .evaluator import TrainsetSchedulingEvaluator
@@ -15,6 +15,8 @@ class GeneticAlgorithmOptimizer:
         self.evaluator = evaluator
         self.config = config or OptimizationConfig()
         self.n_genes = evaluator.num_trainsets
+        self.n_blocks = evaluator.num_blocks
+        self.optimize_blocks = self.config.optimize_block_assignment
         
     def initialize_population(self) -> np.ndarray:
         """Initialize random population with smart seeding."""
@@ -26,7 +28,7 @@ class GeneticAlgorithmOptimizer:
             
             # Randomly assign required number to service, min to standby, rest to maintenance
             indices = np.random.permutation(self.n_genes)
-            service_count = self.config.required_service_trains
+            service_count = min(self.config.required_service_trains, self.n_genes)
             standby_count = self.config.min_standby
             
             solution[indices[:service_count]] = 0  # Service
@@ -42,6 +44,32 @@ class GeneticAlgorithmOptimizer:
         
         return np.array(population)
     
+    def initialize_block_population(self, trainset_population: np.ndarray) -> np.ndarray:
+        """Initialize block assignment population.
+        
+        For each trainset solution, initialize a block assignment solution.
+        block_solution[i] = index of trainset assigned to block i, or -1 if unassigned.
+        """
+        block_population = []
+        
+        for trainset_sol in trainset_population:
+            # Get service train indices
+            service_indices = np.where(trainset_sol == 0)[0]
+            
+            if len(service_indices) == 0:
+                # No service trains, all blocks unassigned
+                block_sol = np.full(self.n_blocks, -1, dtype=int)
+            else:
+                # Distribute blocks evenly among service trains
+                block_sol = np.zeros(self.n_blocks, dtype=int)
+                for i in range(self.n_blocks):
+                    # Assign to a random service train
+                    block_sol[i] = service_indices[i % len(service_indices)]
+            
+            block_population.append(block_sol)
+        
+        return np.array(block_population)
+    
     def tournament_selection(self, population: np.ndarray, 
                             fitness: np.ndarray, 
                             tournament_size: int = 5) -> np.ndarray:
@@ -83,8 +111,9 @@ class GeneticAlgorithmOptimizer:
         standby_count = np.sum(repaired == 1)
         
         # If too few in service, convert some from maintenance/standby
-        if service_count < self.config.required_service_trains:
-            needed = self.config.required_service_trains - service_count
+        target_service = min(self.config.required_service_trains, self.n_genes - self.config.min_standby)
+        if service_count < target_service:
+            needed = target_service - service_count
             candidates = np.where((repaired == 1) | (repaired == 2))[0]
             if len(candidates) >= needed:
                 selected = np.random.choice(candidates, needed, replace=False)
@@ -101,35 +130,83 @@ class GeneticAlgorithmOptimizer:
         
         return repaired
     
+    def repair_block_solution(self, block_sol: np.ndarray, trainset_sol: np.ndarray) -> np.ndarray:
+        """Repair block assignments to only assign to service trains."""
+        repaired = block_sol.copy()
+        service_indices = np.where(trainset_sol == 0)[0]
+        
+        if len(service_indices) == 0:
+            return np.full(self.n_blocks, -1, dtype=int)
+        
+        for i in range(len(repaired)):
+            if repaired[i] not in service_indices:
+                # Reassign to a random service train
+                repaired[i] = np.random.choice(service_indices)
+        
+        return repaired
+    
+    def mutate_block_solution(self, block_sol: np.ndarray, service_indices: np.ndarray) -> np.ndarray:
+        """Mutate block assignments."""
+        mutated = block_sol.copy()
+        
+        if len(service_indices) == 0:
+            return mutated
+        
+        for i in range(len(mutated)):
+            if np.random.random() < self.config.mutation_rate:
+                mutated[i] = np.random.choice(service_indices)
+        
+        return mutated
+    
     def optimize(self) -> OptimizationResult:
         """Run genetic algorithm optimization."""
         population = self.initialize_population()
+        
+        # Initialize block population if optimizing blocks
+        block_population = None
+        if self.optimize_blocks:
+            block_population = self.initialize_block_population(population)
+        
         best_fitness = float('inf')
         best_solution: np.ndarray = population[0].copy()
+        best_block_solution: Optional[np.ndarray] = None
         
         print(f"Starting GA optimization with {self.config.population_size} individuals for {self.config.generations} generations")
+        if self.optimize_blocks:
+            print(f"Optimizing block assignments for {self.n_blocks} service blocks")
         
         for gen in range(self.config.generations):
             try:
                 # Evaluate fitness
-                fitness = np.array([self.evaluator.fitness_function(ind) for ind in population])
+                if self.optimize_blocks and block_population is not None:
+                    fitness = np.array([
+                        self.evaluator.schedule_fitness_function(population[i], block_population[i])
+                        for i in range(len(population))
+                    ])
+                else:
+                    fitness = np.array([self.evaluator.fitness_function(ind) for ind in population])
                 
                 # Track best solution
                 gen_best_idx = np.argmin(fitness)
                 if fitness[gen_best_idx] < best_fitness:
                     best_fitness = fitness[gen_best_idx]
                     best_solution = population[gen_best_idx].copy()
+                    if self.optimize_blocks and block_population is not None:
+                        best_block_solution = block_population[gen_best_idx].copy()
                 
                 if gen % 50 == 0:
                     print(f"Generation {gen}: Best Fitness = {best_fitness:.2f}")
                 
                 # Create new population
                 new_population = []
+                new_block_population = [] if self.optimize_blocks else None
                 
                 # Elitism - keep best solutions
                 elite_indices = np.argsort(fitness)[:self.config.elite_size]
                 for idx in elite_indices:
                     new_population.append(population[idx].copy())
+                    if self.optimize_blocks and block_population is not None:
+                        new_block_population.append(block_population[idx].copy())
                 
                 # Generate offspring through selection, crossover, and mutation
                 while len(new_population) < self.config.population_size:
@@ -147,17 +224,49 @@ class GeneticAlgorithmOptimizer:
                     new_population.append(child1)
                     if len(new_population) < self.config.population_size:
                         new_population.append(child2)
+                    
+                    # Handle block solutions
+                    if self.optimize_blocks and new_block_population is not None:
+                        service_indices_1 = np.where(child1 == 0)[0]
+                        service_indices_2 = np.where(child2 == 0)[0]
+                        
+                        # Create new block assignments for children
+                        block_child1 = self._create_block_for_trainset(child1)
+                        block_child1 = self.mutate_block_solution(block_child1, service_indices_1)
+                        
+                        new_block_population.append(block_child1)
+                        
+                        if len(new_block_population) < self.config.population_size:
+                            block_child2 = self._create_block_for_trainset(child2)
+                            block_child2 = self.mutate_block_solution(block_child2, service_indices_2)
+                            new_block_population.append(block_child2)
                 
                 population = np.array(new_population)
+                if self.optimize_blocks:
+                    block_population = np.array(new_block_population)
                 
             except Exception as e:
                 print(f"Error in generation {gen}: {e}")
                 break
         
         # Build result
-        return self._build_result(best_solution, best_fitness)
+        return self._build_result(best_solution, best_fitness, best_block_solution)
     
-    def _build_result(self, solution: np.ndarray, fitness: float) -> OptimizationResult:
+    def _create_block_for_trainset(self, trainset_sol: np.ndarray) -> np.ndarray:
+        """Create block assignments for a trainset solution."""
+        service_indices = np.where(trainset_sol == 0)[0]
+        
+        if len(service_indices) == 0:
+            return np.full(self.n_blocks, -1, dtype=int)
+        
+        block_sol = np.zeros(self.n_blocks, dtype=int)
+        for i in range(self.n_blocks):
+            block_sol[i] = service_indices[i % len(service_indices)]
+        
+        return block_sol
+    
+    def _build_result(self, solution: np.ndarray, fitness: float, 
+                      block_solution: Optional[np.ndarray] = None) -> OptimizationResult:
         """Build optimization result from solution."""
         objectives = self.evaluator.calculate_objectives(solution)
         
@@ -172,11 +281,25 @@ class GeneticAlgorithmOptimizer:
             valid, reason = self.evaluator.check_hard_constraints(ts_id)
             explanations[ts_id] = "�� Fit for service" if valid else f"⚠ {reason}"
         
+        # Build block assignments
+        block_assignments = {}
+        if block_solution is not None and self.optimize_blocks:
+            for ts_id in service:
+                block_assignments[ts_id] = []
+            
+            for block_idx, train_idx in enumerate(block_solution):
+                if 0 <= train_idx < len(self.evaluator.trainsets):
+                    ts_id = self.evaluator.trainsets[train_idx]
+                    if ts_id in block_assignments:
+                        block_id = self.evaluator.all_blocks[block_idx]['block_id']
+                        block_assignments[ts_id].append(block_id)
+        
         return OptimizationResult(
             selected_trainsets=service,
             standby_trainsets=standby,
             maintenance_trainsets=maintenance,
             objectives=objectives,
             fitness_score=fitness,
-            explanation=explanations
+            explanation=explanations,
+            service_block_assignments=block_assignments
         )

greedyOptim/models.py

@@ -197,6 +197,8 @@ class OptimizationResult:
     objectives: Dict[str, float]
     fitness_score: float
     explanation: Dict[str, str]
+    # Block assignments: maps trainset_id -> list of block_ids
+    service_block_assignments: Dict[str, List[str]] = field(default_factory=dict)
 
 
 @dataclass
@@ -206,9 +208,11 @@ class OptimizationConfig:
     min_standby: int = 2
     population_size: int = 100
     generations: int = 200
+    iterations: int = 15  # For SA, CMA-ES, PSO (configurable)
     mutation_rate: float = 0.1
     crossover_rate: float = 0.8
     elite_size: int = 5
+    optimize_block_assignment: bool = True  # Enable block assignment optimization
 
 
 @dataclass

greedyOptim/schedule_generator.py

@@ -94,11 +94,30 @@ class ScheduleGenerator:
         total_km = 0
         mileages = []
         
+        # Check if we have optimized block assignments
+        has_optimized_blocks = (
+            optimization_result.service_block_assignments and 
+            len(optimization_result.service_block_assignments) > 0
+        )
+        
+        # Build block lookup for optimized assignments
+        block_lookup = {}
+        if has_optimized_blocks:
+            all_blocks = self.service_block_generator.get_all_service_blocks()
+            block_lookup = {b['block_id']: b for b in all_blocks}
+        
         # Service trains
         num_service = len(optimization_result.selected_trainsets)
         for idx, ts_id in enumerate(optimization_result.selected_trainsets):
+            # Get optimized blocks for this trainset if available
+            assigned_block_ids = None
+            if has_optimized_blocks:
+                assigned_block_ids = optimization_result.service_block_assignments.get(ts_id, [])
+            
             trainset, ts_alerts, km = self._generate_service_trainset(
-                ts_id, idx, num_service
+                ts_id, idx, num_service, 
+                assigned_block_ids=assigned_block_ids,
+                block_lookup=block_lookup
             )
             trainsets.append(trainset)
             alerts.extend(ts_alerts)
@@ -163,18 +182,38 @@ class ScheduleGenerator:
         self,
         trainset_id: str,
         index: int,
-        num_service: int
+        num_service: int,
+        assigned_block_ids: list = None,
+        block_lookup: dict = None
     ) -> tuple:
         """Generate schedule for a service trainset.
         
+        Args:
+            trainset_id: ID of the trainset
+            index: Index in service trainsets list
+            num_service: Total number of service trainsets
+            assigned_block_ids: List of block IDs from optimizer (if using optimized blocks)
+            block_lookup: Dictionary mapping block_id to block data
+        
         Returns:
             Tuple of (ScheduleTrainset, alerts, daily_km)
         """
         ts_data = self.status_map.get(trainset_id, {})
         cumulative_km = ts_data.get('total_mileage_km', 0)
         
-        # Generate service blocks
-        blocks_data = self.service_block_generator.generate_service_blocks(index, num_service)
+        # Use optimized blocks if available, otherwise fall back to index-based generation
+        if assigned_block_ids and block_lookup:
+            # Use optimizer-assigned blocks
+            blocks_data = []
+            for block_id in assigned_block_ids:
+                if block_id in block_lookup:
+                    blocks_data.append(block_lookup[block_id])
+            # Sort by departure time
+            blocks_data.sort(key=lambda b: b['departure_time'])
+        else:
+            # Fall back to legacy index-based block generation
+            blocks_data = self.service_block_generator.generate_service_blocks(index, num_service)
+        
         service_blocks = [
             ServiceBlock(
                 block_id=b['block_id'],

greedyOptim/scheduler.py

@@ -69,15 +69,16 @@ class TrainsetSchedulingOptimizer:
                 result = optimizer.optimize()
             elif method == 'cmaes':
                 optimizer = CMAESOptimizer(self.evaluator, self.config)
-                generations = kwargs.get('generations', 150)
+                # Use config.iterations if generations not specified in kwargs
+                generations = kwargs.get('generations')  # None means use optimizer's default
                 result = optimizer.optimize(generations)
             elif method == 'pso':
                 optimizer = ParticleSwarmOptimizer(self.evaluator, self.config)
-                generations = kwargs.get('generations', 200)
+                generations = kwargs.get('generations')  # None means use optimizer's default
                 result = optimizer.optimize(generations)
             elif method == 'sa':
                 optimizer = SimulatedAnnealingOptimizer(self.evaluator, self.config)
-                max_iterations = kwargs.get('max_iterations', 10000)
+                max_iterations = kwargs.get('max_iterations')  # None means use optimizer's default
                 result = optimizer.optimize(max_iterations)
             elif method == 'nsga2':
                 optimizer = MultiObjectiveOptimizer(self.evaluator, self.config)

greedyOptim/service_blocks.py

@@ -2,8 +2,8 @@
 Service Block Generator
 Generates realistic service blocks with departure times for train schedules.
 """
-from typing import List, Dict
-from datetime import time
+from typing import List, Dict, Tuple
+from datetime import time, datetime, timedelta
 
 
 class ServiceBlockGenerator:
@@ -27,6 +27,112 @@ class ServiceBlockGenerator:
         """Initialize service block generator."""
         self.round_trip_time_hours = (self.ROUTE_LENGTH_KM * 2) / self.AVG_SPEED_KMH
         self.round_trip_time_minutes = self.round_trip_time_hours * 60
+        self._all_blocks_cache = None
+    
+    def get_all_service_blocks(self) -> List[Dict]:
+        """Get all available service blocks for the day.
+        
+        Pre-generates all possible service blocks that need to be assigned to trainsets.
+        These represent the "slots" that the optimizer will fill.
+        
+        Returns:
+            List of all service block dictionaries with block_id, departure_time, etc.
+        """
+        if self._all_blocks_cache is not None:
+            return self._all_blocks_cache
+        
+        all_blocks = []
+        block_counter = 0
+        
+        # Morning peak blocks (7:00 - 10:00)
+        # Need departures every 6 minutes = 10 per hour = 30 blocks
+        for hour in [7, 8, 9]:
+            for minute in range(0, 60, 6):
+                block_counter += 1
+                origin = self.TERMINALS[block_counter % 2]
+                destination = self.TERMINALS[(block_counter + 1) % 2]
+                all_blocks.append({
+                    'block_id': f'BLK-{block_counter:03d}',
+                    'departure_time': f'{hour:02d}:{minute:02d}',
+                    'origin': origin,
+                    'destination': destination,
+                    'trip_count': 3,  # ~3 hours of peak service
+                    'estimated_km': int(3 * self.ROUTE_LENGTH_KM * 2),
+                    'period': 'morning_peak',
+                    'is_peak': True
+                })
+        
+        # Midday off-peak blocks (10:00 - 17:00)
+        # Need departures every 15 minutes = 4 per hour = 28 blocks
+        for hour in range(10, 17):
+            for minute in range(0, 60, 15):
+                block_counter += 1
+                origin = self.TERMINALS[block_counter % 2]
+                destination = self.TERMINALS[(block_counter + 1) % 2]
+                all_blocks.append({
+                    'block_id': f'BLK-{block_counter:03d}',
+                    'departure_time': f'{hour:02d}:{minute:02d}',
+                    'origin': origin,
+                    'destination': destination,
+                    'trip_count': 2,
+                    'estimated_km': int(2 * self.ROUTE_LENGTH_KM * 2),
+                    'period': 'midday',
+                    'is_peak': False
+                })
+        
+        # Evening peak blocks (17:00 - 21:00)
+        # Need departures every 6 minutes = 10 per hour = 40 blocks
+        for hour in range(17, 21):
+            for minute in range(0, 60, 6):
+                block_counter += 1
+                origin = self.TERMINALS[block_counter % 2]
+                destination = self.TERMINALS[(block_counter + 1) % 2]
+                all_blocks.append({
+                    'block_id': f'BLK-{block_counter:03d}',
+                    'departure_time': f'{hour:02d}:{minute:02d}',
+                    'origin': origin,
+                    'destination': destination,
+                    'trip_count': 3,
+                    'estimated_km': int(3 * self.ROUTE_LENGTH_KM * 2),
+                    'period': 'evening_peak',
+                    'is_peak': True
+                })
+        
+        # Late evening blocks (21:00 - 23:00)
+        # Need departures every 15 minutes = 4 per hour = 8 blocks
+        for hour in range(21, 23):
+            for minute in range(0, 60, 15):
+                block_counter += 1
+                origin = self.TERMINALS[block_counter % 2]
+                destination = self.TERMINALS[(block_counter + 1) % 2]
+                all_blocks.append({
+                    'block_id': f'BLK-{block_counter:03d}',
+                    'departure_time': f'{hour:02d}:{minute:02d}',
+                    'origin': origin,
+                    'destination': destination,
+                    'trip_count': 1,
+                    'estimated_km': int(1 * self.ROUTE_LENGTH_KM * 2),
+                    'period': 'late_evening',
+                    'is_peak': False
+                })
+        
+        self._all_blocks_cache = all_blocks
+        return all_blocks
+    
+    def get_block_count(self) -> int:
+        """Get total number of service blocks."""
+        return len(self.get_all_service_blocks())
+    
+    def get_peak_block_indices(self) -> List[int]:
+        """Get indices of peak hour blocks."""
+        blocks = self.get_all_service_blocks()
+        return [i for i, b in enumerate(blocks) if b['is_peak']]
+    
+    def get_blocks_by_ids(self, block_ids: List[str]) -> List[Dict]:
+        """Get blocks by their IDs."""
+        all_blocks = self.get_all_service_blocks()
+        block_map = {b['block_id']: b for b in all_blocks}
+        return [block_map[bid] for bid in block_ids if bid in block_map]
     
     def generate_service_blocks(self, train_index: int, num_service_trains: int) -> List[Dict]:
         """Generate service blocks for a train with staggered departures.