File size: 8,987 Bytes
33c9f7f 1b2c6dc f6aa8e8 1b2c6dc f6aa8e8 1b2c6dc f6aa8e8 1b2c6dc f6aa8e8 1b2c6dc 33c9f7f 0f86bb9 33c9f7f 0f86bb9 33c9f7f 0f86bb9 33c9f7f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 | """
Data models and dataclasses for the optimization system.
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
class TrainStatus(str, Enum):
"""Train operational status for schedule output."""
REVENUE_SERVICE = "REVENUE_SERVICE"
STANDBY = "STANDBY"
MAINTENANCE = "MAINTENANCE"
OUT_OF_SERVICE = "OUT_OF_SERVICE"
class MaintenanceType(str, Enum):
"""Types of maintenance."""
SCHEDULED_INSPECTION = "SCHEDULED_INSPECTION"
PREVENTIVE = "PREVENTIVE"
CORRECTIVE = "CORRECTIVE"
class AlertSeverity(str, Enum):
"""Alert severity levels."""
LOW = "LOW"
MEDIUM = "MEDIUM"
HIGH = "HIGH"
CRITICAL = "CRITICAL"
@dataclass
class StationStop:
"""A single station stop within a trip."""
station_code: str
station_name: str
arrival_time: Optional[str] # HH:MM format, None for origin
departure_time: Optional[str] # HH:MM format, None for destination
distance_from_origin_km: float
platform: Optional[int] = None
def to_dict(self) -> Dict:
result = {
'station_code': self.station_code,
'station_name': self.station_name,
'distance_from_origin_km': self.distance_from_origin_km
}
if self.arrival_time:
result['arrival_time'] = self.arrival_time
if self.departure_time:
result['departure_time'] = self.departure_time
if self.platform:
result['platform'] = self.platform
return result
@dataclass
class Trip:
"""A single trip from origin to destination with all stops."""
trip_id: str
trip_number: int # 1, 2, 3... within the block
direction: str # "UP" (towards Pettah) or "DOWN" (towards Aluva)
origin: str
destination: str
departure_time: str
arrival_time: str
stops: List[StationStop] = field(default_factory=list)
def to_dict(self) -> Dict:
return {
'trip_id': self.trip_id,
'trip_number': self.trip_number,
'direction': self.direction,
'origin': self.origin,
'destination': self.destination,
'departure_time': self.departure_time,
'arrival_time': self.arrival_time,
'stops': [s.to_dict() for s in self.stops]
}
@dataclass
class ServiceBlock:
"""A service block represents a continuous operating period for a train."""
block_id: str
departure_time: str # HH:MM format
origin: str
destination: str
trip_count: int
estimated_km: int
# Enhanced fields
journey_time_minutes: Optional[float] = None
trips: List[Trip] = field(default_factory=list) # Detailed trip breakdown
period: Optional[str] = None # morning_peak, midday, evening_peak, late_evening
is_peak: bool = False
def to_dict(self) -> Dict:
result = {
'block_id': self.block_id,
'departure_time': self.departure_time,
'origin': self.origin,
'destination': self.destination,
'trip_count': self.trip_count,
'estimated_km': self.estimated_km
}
if self.journey_time_minutes:
result['journey_time_minutes'] = self.journey_time_minutes
if self.period:
result['period'] = self.period
result['is_peak'] = self.is_peak
if self.trips:
result['trips'] = [t.to_dict() for t in self.trips]
return result
@dataclass
class ScheduleTrainset:
"""Complete trainset information in the schedule."""
trainset_id: str
status: TrainStatus
readiness_score: float
daily_km_allocation: int
cumulative_km: int
# For REVENUE_SERVICE
assigned_duty: Optional[str] = None
priority_rank: Optional[int] = None
service_blocks: List[ServiceBlock] = field(default_factory=list)
stabling_bay: Optional[str] = None
# For STANDBY
standby_reason: Optional[str] = None
# For MAINTENANCE
maintenance_type: Optional[MaintenanceType] = None
ibl_bay: Optional[str] = None
estimated_completion: Optional[str] = None
# Alerts
alerts: List[str] = field(default_factory=list)
def to_dict(self) -> Dict:
result = {
'trainset_id': self.trainset_id,
'status': self.status.value,
'readiness_score': self.readiness_score,
'daily_km_allocation': self.daily_km_allocation,
'cumulative_km': self.cumulative_km,
'alerts': self.alerts
}
if self.assigned_duty:
result['assigned_duty'] = self.assigned_duty
if self.priority_rank is not None:
result['priority_rank'] = self.priority_rank
if self.service_blocks:
result['service_blocks'] = [b.to_dict() for b in self.service_blocks]
if self.stabling_bay:
result['stabling_bay'] = self.stabling_bay
if self.standby_reason:
result['standby_reason'] = self.standby_reason
if self.maintenance_type:
result['maintenance_type'] = self.maintenance_type.value
if self.ibl_bay:
result['ibl_bay'] = self.ibl_bay
if self.estimated_completion:
result['estimated_completion'] = self.estimated_completion
return result
@dataclass
class FleetSummary:
"""Summary statistics for the fleet."""
total_trainsets: int
revenue_service: int
standby: int
maintenance: int
availability_percent: float
def to_dict(self) -> Dict:
return {
'total_trainsets': self.total_trainsets,
'revenue_service': self.revenue_service,
'standby': self.standby,
'maintenance': self.maintenance,
'availability_percent': self.availability_percent
}
@dataclass
class ScheduleAlert:
"""Alert or warning in the schedule."""
trainset_id: str
severity: AlertSeverity
alert_type: str
message: str
def to_dict(self) -> Dict:
return {
'trainset_id': self.trainset_id,
'severity': self.severity.value,
'alert_type': self.alert_type,
'message': self.message
}
@dataclass
class OptimizationMetrics:
"""Metrics about the optimization."""
fitness_score: float
method: str
mileage_variance_coefficient: float
total_planned_km: int
optimization_runtime_ms: int
def to_dict(self) -> Dict:
return {
'fitness_score': self.fitness_score,
'method': self.method,
'mileage_variance_coefficient': self.mileage_variance_coefficient,
'total_planned_km': self.total_planned_km,
'optimization_runtime_ms': self.optimization_runtime_ms
}
@dataclass
class ScheduleResult:
"""Complete schedule result with all trainsets and service blocks."""
schedule_id: str
generated_at: str
valid_from: str
valid_until: str
depot: str
trainsets: List[ScheduleTrainset]
fleet_summary: FleetSummary
optimization_metrics: OptimizationMetrics
alerts: List[ScheduleAlert] = field(default_factory=list)
def to_dict(self) -> Dict:
"""Convert to dictionary for JSON serialization and benchmarks."""
return {
'schedule_id': self.schedule_id,
'generated_at': self.generated_at,
'valid_from': self.valid_from,
'valid_until': self.valid_until,
'depot': self.depot,
'trainsets': [ts.to_dict() for ts in self.trainsets],
'fleet_summary': self.fleet_summary.to_dict(),
'optimization_metrics': self.optimization_metrics.to_dict(),
'alerts': [a.to_dict() for a in self.alerts]
}
@dataclass
class OptimizationResult:
"""Result of a trainset scheduling optimization run."""
selected_trainsets: List[str]
standby_trainsets: List[str]
maintenance_trainsets: List[str]
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
class OptimizationConfig:
"""Configuration parameters for optimization algorithms."""
required_service_trains: int = 20
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
class TrainsetConstraints:
"""Hard and soft constraints for a trainset."""
has_valid_certificates: bool
has_critical_jobs: bool
component_warnings: List[str]
maintenance_due: bool
mileage: int
last_service_days: int |