120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
"""Hard constraints - must always be satisfied."""
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Any, Optional
|
|
from constraints.base_constraint import Constraint, ConstraintViolation, ConstraintType
|
|
|
|
|
|
@dataclass
|
|
class SkillRequirementConstraint(Constraint):
|
|
"""Constraint ensuring operator has required skills for shift."""
|
|
operator_id: str
|
|
required_skills: List[str]
|
|
shift_id: str
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if operator has all required skills."""
|
|
operator_skills = context.get('operators', {}).get(self.operator_id, {}).get('skills', [])
|
|
return all(skill in operator_skills for skill in self.required_skills)
|
|
|
|
|
|
@dataclass
|
|
class ShiftCoverageConstraint(Constraint):
|
|
"""Constraint ensuring shift is covered by at least one operator."""
|
|
shift_id: str
|
|
minimum_coverage: int = 1
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if shift has minimum coverage."""
|
|
assigned_operators = context.get('assignments', {}).get(self.shift_id, [])
|
|
return len(assigned_operators) >= self.minimum_coverage
|
|
|
|
|
|
@dataclass
|
|
class MaxWorkTimeConstraint(Constraint):
|
|
"""Constraint limiting maximum working hours per day/week."""
|
|
operator_id: str
|
|
max_daily_hours: int = 8
|
|
max_weekly_hours: int = 40
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if operator hasn't exceeded max hours."""
|
|
operator_shifts = context.get('operators', {}).get(self.operator_id, {}).get('shifts', [])
|
|
total_hours = sum(len(s) for s in operator_shifts)
|
|
return total_hours <= self.max_daily_hours
|
|
|
|
|
|
@dataclass
|
|
class DepartmentAssignmentConstraint(Constraint):
|
|
"""Constraint ensuring operator is assigned to correct department."""
|
|
operator_id: str
|
|
allowed_departments: List[str]
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if operator is in allowed department."""
|
|
assigned_dept = context.get('assignments', {}).get(self.operator_id, {}).get('department')
|
|
return assigned_dept in self.allowed_departments
|
|
|
|
|
|
@dataclass
|
|
class MinCoverageConstraint(Constraint):
|
|
"""Constraint ensuring minimum operators per shift."""
|
|
shift_id: str
|
|
min_operators: int
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check if minimum operators are assigned."""
|
|
assigned = context.get('assignments', {}).get(self.shift_id, [])
|
|
return len(assigned) >= self.min_operators
|
|
|
|
|
|
@dataclass
|
|
class ConflictPreventionConstraint(Constraint):
|
|
"""Constraint preventing conflicting assignments."""
|
|
operator_id: str
|
|
conflicts: List[str]
|
|
|
|
def is_satisfied(self, context: Dict[str, Any]) -> bool:
|
|
"""Check for conflicts."""
|
|
current_conflicts = [c for c in self.conflicts if c in context.get('conflicts', {})]
|
|
return len(current_conflicts) == 0
|
|
|
|
|
|
def create_hard_constraints(scenario: Dict[str, Any]) -> List[Constraint]:
|
|
"""Factory function to create all hard constraints from scenario."""
|
|
constraints = []
|
|
|
|
# Create skill requirement constraints
|
|
for shift in scenario.get('events', []):
|
|
for agent in scenario.get('agents', []):
|
|
if agent.get('type') == 'operator':
|
|
constraints.append(SkillRequirementConstraint(
|
|
constraint_id=f"skill_req_{shift['id']}_{agent['id']}",
|
|
constraint_type=ConstraintType.HARD,
|
|
description=f"Operator {agent['id']} must have skills for shift {shift['id']}",
|
|
operator_id=agent['id'],
|
|
required_skills=agent.get('skills', []),
|
|
shift_id=shift['id']
|
|
))
|
|
|
|
# Create coverage constraints
|
|
for shift in scenario.get('events', []):
|
|
constraints.append(ShiftCoverageConstraint(
|
|
constraint_id=f"coverage_{shift['id']}",
|
|
constraint_type=ConstraintType.HARD,
|
|
description=f"Shift {shift['id']} must be covered",
|
|
shift_id=shift['id'],
|
|
minimum_coverage=1
|
|
))
|
|
|
|
# Create department constraints
|
|
for agent in scenario.get('agents', []):
|
|
if agent.get('type') == 'operator':
|
|
constraints.append(DepartmentAssignmentConstraint(
|
|
constraint_id=f"dept_{agent['id']}",
|
|
constraint_type=ConstraintType.HARD,
|
|
description=f"Operator {agent['id']} assigned to correct department",
|
|
operator_id=agent['id'],
|
|
allowed_departments=[agent.get('department', '')]
|
|
))
|
|
|
|
return constraints |