123 lines
5.0 KiB
Python
123 lines
5.0 KiB
Python
"""Soft constraints - should be satisfied when possible."""
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Any, Optional
|
|
from constraints.base_constraint import Constraint, ConstraintViolation, ConstraintType
|
|
|
|
|
|
@dataclass
|
|
class OperatorPreferenceConstraint(Constraint):
|
|
"""Soft constraint for operator shift preferences."""
|
|
operator_id: str
|
|
preferred_shifts: List[str] # Shift types operator prefers
|
|
priority: int = 1 # Higher = more important preference
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if operator is assigned preferred shifts."""
|
|
assigned_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', [])
|
|
assigned_shift_types = [s.get('shift_type') for s in assigned_shifts]
|
|
return all(pref in assigned_shift_types for pref in self.preferred_shifts)
|
|
|
|
|
|
@dataclass
|
|
class DayAvailabilityConstraint(Constraint):
|
|
"""Soft constraint for operator day availability."""
|
|
operator_id: str
|
|
preferred_days: List[str]
|
|
priority: int = 1
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if operator works on preferred days."""
|
|
assigned_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', [])
|
|
assigned_days = set()
|
|
for shift in assigned_shifts:
|
|
day_name = shift.get('date', {}).get('strftime', '%Y-%m-%d')
|
|
assigned_days.add(day_name)
|
|
return all(day in assigned_days for day in self.preferred_days)
|
|
|
|
|
|
@dataclass
|
|
class WorkBalanceConstraint(Constraint):
|
|
"""Soft constraint for balanced workload distribution."""
|
|
shift_id: str
|
|
max_assignments: int = 3 # Don't assign more than X people to same shift
|
|
priority: int = 2
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if shift isn't overloaded."""
|
|
assigned = context.get('assignments', {}).get(self.shift_id, [])
|
|
return len(assigned) <= self.max_assignments
|
|
|
|
|
|
@dataclass
|
|
class ContinuityPreference(Constraint):
|
|
"""Soft constraint for consecutive shift preferences."""
|
|
operator_id: str
|
|
preferred_pattern: str # e.g., "morning,afternoon" for consecutive shifts
|
|
priority: int = 1
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if consecutive shifts match preference."""
|
|
assigned_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', [])
|
|
if len(assigned_shifts) < 2:
|
|
return True
|
|
# Check if shifts are consecutive and match pattern
|
|
return True # Implementation depends on shift ordering
|
|
|
|
|
|
@dataclass
|
|
class SeniorityPriorityConstraint(Constraint):
|
|
"""Soft constraint to prioritize senior operators."""
|
|
shift_id: str
|
|
max_junior_assignments: int = 2
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if enough senior operators are assigned."""
|
|
assigned = context.get('assignments', {}).get(self.shift_id, [])
|
|
junior_count = sum(1 for op in assigned if op.get('seniority', 0) < 3)
|
|
return junior_count <= self.max_junior_assignments
|
|
|
|
|
|
def create_soft_constraints(scenario: Dict[str, Any]) -> List[Constraint]:
|
|
"""Factory function to create all soft constraints from scenario."""
|
|
constraints = []
|
|
|
|
# Create preference constraints
|
|
for agent in scenario.get('agents', []):
|
|
if agent.get('type') == 'operator':
|
|
prefs = agent.get('preferences', {})
|
|
if prefs:
|
|
constraints.append(OperatorPreferenceConstraint(
|
|
constraint_id=f"pref_{agent['id']}",
|
|
constraint_type=ConstraintType.SOFT,
|
|
description=f"Operator {agent['id']} prefers shifts: {prefs}",
|
|
operator_id=agent['id'],
|
|
preferred_shifts=list(prefs.keys()) if isinstance(prefs, dict) else prefs,
|
|
priority=1
|
|
))
|
|
|
|
# Create day availability constraints
|
|
for agent in scenario.get('agents', []):
|
|
if agent.get('type') == 'operator':
|
|
preferred_days = agent.get('preferred_days', [])
|
|
if preferred_days:
|
|
constraints.append(DayAvailabilityConstraint(
|
|
constraint_id=f"day_pref_{agent['id']}",
|
|
constraint_type=ConstraintType.SOFT,
|
|
description=f"Operator {agent['id']} prefers days: {preferred_days}",
|
|
operator_id=agent['id'],
|
|
preferred_days=preferred_days,
|
|
priority=1
|
|
))
|
|
|
|
# Create work balance constraints
|
|
for shift in scenario.get('events', []):
|
|
constraints.append(WorkBalanceConstraint(
|
|
constraint_id=f"balance_{shift['id']}",
|
|
constraint_type=ConstraintType.SOFT,
|
|
description=f"Limit assignments for shift {shift['id']}",
|
|
shift_id=shift['id'],
|
|
max_assignments=scenario.get('constraints', {}).get('max_per_shift', 3),
|
|
priority=2
|
|
))
|
|
|
|
return constraints |