"""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