Spaces:
Paused
Paused
| from typing import Dict, List, Any | |
| from constraint_solvers.timetable.domain import EmployeeSchedule | |
| from constraint_solvers.timetable.solver import solution_manager | |
| class ConstraintAnalyzerService: | |
| """ | |
| Service for analyzing scheduling solutions using Timefold's native constraint analysis. | |
| This service provides methods to analyze constraint violations, generate suggestions, | |
| and understand solution quality using Timefold's built-in ScoreAnalysis and SolutionManager APIs. | |
| """ | |
| def analyze_constraint_violations(schedule: EmployeeSchedule) -> str: | |
| """ | |
| Analyze constraint violations in a schedule using Timefold's native score analysis. | |
| Args: | |
| schedule: The schedule to analyze | |
| Returns: | |
| Detailed string describing constraint violations and their breakdown | |
| """ | |
| if not schedule.score or schedule.score.hard_score >= 0: | |
| return "No constraint violations detected." | |
| # Get Timefold's solution manager and analyze the schedule | |
| score_analysis = solution_manager.analyze(schedule) | |
| # Return the built-in summary | |
| return score_analysis.summary | |
| def get_detailed_analysis(schedule: EmployeeSchedule) -> Dict[str, Any]: | |
| """ | |
| Get detailed constraint analysis as a structured dictionary. | |
| Args: | |
| schedule: The schedule to analyze | |
| Returns: | |
| Dictionary containing detailed constraint analysis information | |
| """ | |
| score_analysis = solution_manager.analyze(schedule) | |
| analysis_result = { | |
| "total_score": str(score_analysis.score), | |
| "hard_score": score_analysis.score.hard_score, | |
| "soft_score": score_analysis.score.soft_score, | |
| "constraints": {}, | |
| } | |
| # Analyze each constraint | |
| for ( | |
| constraint_ref, | |
| constraint_analysis, | |
| ) in score_analysis.constraint_map.items(): | |
| constraint_id = constraint_ref.constraint_id | |
| constraint_info = { | |
| "score": str(constraint_analysis.score), | |
| "match_count": constraint_analysis.match_count, | |
| "matches": [], | |
| } | |
| # Get details for each constraint match | |
| for match_analysis in constraint_analysis.matches: | |
| match_info = { | |
| "score": str(match_analysis.score), | |
| "justification": str(match_analysis.justification) | |
| if match_analysis.justification | |
| else None, | |
| } | |
| constraint_info["matches"].append(match_info) | |
| analysis_result["constraints"][constraint_id] = constraint_info | |
| return analysis_result | |
| def get_broken_constraints(schedule: EmployeeSchedule) -> List[Dict[str, Any]]: | |
| """ | |
| Get a list of broken constraints with their details. | |
| Args: | |
| schedule: The schedule to analyze | |
| Returns: | |
| List of dictionaries, each containing information about a broken constraint | |
| """ | |
| score_analysis = solution_manager.analyze(schedule) | |
| broken_constraints = [] | |
| for ( | |
| constraint_ref, | |
| constraint_analysis, | |
| ) in score_analysis.constraint_map.items(): | |
| # Only include constraints that have a negative impact on the score | |
| if ( | |
| constraint_analysis.score.hard_score < 0 | |
| or constraint_analysis.score.soft_score < 0 | |
| ): | |
| broken_constraints.append( | |
| { | |
| "constraint_id": constraint_ref.constraint_id, | |
| "score": str(constraint_analysis.score), | |
| "hard_score": constraint_analysis.score.hard_score, | |
| "soft_score": constraint_analysis.score.soft_score, | |
| "match_count": constraint_analysis.match_count, | |
| "constraint_name": constraint_ref.constraint_name, | |
| } | |
| ) | |
| return broken_constraints | |
| def compare_solutions( | |
| old_schedule: EmployeeSchedule, new_schedule: EmployeeSchedule | |
| ) -> Dict[str, Any]: | |
| """ | |
| Compare two solutions and identify what changed between them. | |
| Args: | |
| old_schedule: The previous schedule solution | |
| new_schedule: The new schedule solution | |
| Returns: | |
| Dictionary containing the differences between the two solutions | |
| """ | |
| old_analysis = solution_manager.analyze(old_schedule) | |
| new_analysis = solution_manager.analyze(new_schedule) | |
| # Calculate the difference | |
| diff = old_analysis - new_analysis | |
| comparison_result = { | |
| "old_score": str(old_analysis.score), | |
| "new_score": str(new_analysis.score), | |
| "score_difference": str(diff.score), | |
| "improved": ( | |
| new_analysis.score.hard_score > old_analysis.score.hard_score | |
| or ( | |
| new_analysis.score.hard_score == old_analysis.score.hard_score | |
| and new_analysis.score.soft_score > old_analysis.score.soft_score | |
| ) | |
| ), | |
| "changed_constraints": {}, | |
| } | |
| # Analyze changes in constraints | |
| for constraint_ref, constraint_analysis in diff.constraint_map.items(): | |
| comparison_result["changed_constraints"][constraint_ref.constraint_id] = { | |
| "score_difference": str(constraint_analysis.score), | |
| "match_count": constraint_analysis.match_count, | |
| "changes": [ | |
| str(match.justification) | |
| for match in constraint_analysis.matches | |
| if match.justification | |
| ], | |
| } | |
| return comparison_result | |
| def get_heat_map_data(schedule: EmployeeSchedule) -> Dict[Any, Dict[str, Any]]: | |
| """ | |
| Get heat map data showing which planning entities have the most constraint violations. | |
| Args: | |
| schedule: The schedule to analyze | |
| Returns: | |
| Dictionary mapping planning entities to their constraint impact | |
| """ | |
| score_explanation = solution_manager.explain(schedule) | |
| indictment_map = score_explanation.indictment_map | |
| heat_map_data = {} | |
| # Process indictments for tasks | |
| for task in schedule.tasks: | |
| indictment = indictment_map.get(task) | |
| if indictment is not None: | |
| heat_map_data[task] = { | |
| "total_score": str(indictment.score), | |
| "hard_score": indictment.score.hard_score, | |
| "soft_score": indictment.score.soft_score, | |
| "constraint_matches": [ | |
| { | |
| "constraint_name": match.constraint_name, | |
| "score": str(match.score), | |
| } | |
| for match in indictment.constraint_match_set | |
| ], | |
| } | |
| # Process indictments for employees | |
| for employee in schedule.employees: | |
| indictment = indictment_map.get(employee) | |
| if indictment is not None: | |
| heat_map_data[employee] = { | |
| "total_score": str(indictment.score), | |
| "hard_score": indictment.score.hard_score, | |
| "soft_score": indictment.score.soft_score, | |
| "constraint_matches": [ | |
| { | |
| "constraint_name": match.constraint_name, | |
| "score": str(match.score), | |
| } | |
| for match in indictment.constraint_match_set | |
| ], | |
| } | |
| return heat_map_data | |
| def generate_improvement_suggestions(schedule: EmployeeSchedule) -> List[str]: | |
| """ | |
| Generate improvement suggestions based on constraint analysis. | |
| Args: | |
| schedule: The schedule to analyze | |
| Returns: | |
| List of actionable suggestions for improving the schedule | |
| """ | |
| if not schedule.score or schedule.score.hard_score >= 0: | |
| return [ | |
| "Schedule is feasible. Consider optimizing soft constraints for better quality." | |
| ] | |
| broken_constraints = ConstraintAnalyzerService.get_broken_constraints(schedule) | |
| suggestions = [] | |
| # Generate suggestions based on broken constraint types | |
| for constraint in broken_constraints: | |
| constraint_id = constraint["constraint_id"].lower() | |
| match constraint_id: | |
| case constraint_id if "skill" in constraint_id: | |
| suggestions.append( | |
| f"Skill constraint violation: Consider adding employees with required skills " | |
| f"or reassigning tasks ({constraint['match_count']} violations)" | |
| ) | |
| case constraint_id if "availability" in constraint_id or "time" in constraint_id: | |
| suggestions.append( | |
| f"Time/Availability constraint violation: Check employee schedules and " | |
| f"task timing ({constraint['match_count']} violations)" | |
| ) | |
| case constraint_id if "sequence" in constraint_id or "order" in constraint_id: | |
| suggestions.append( | |
| f"Sequencing constraint violation: Review task dependencies and ordering " | |
| f"({constraint['match_count']} violations)" | |
| ) | |
| case constraint_id if "capacity" in constraint_id or "workload" in constraint_id: | |
| suggestions.append( | |
| f"Capacity constraint violation: Distribute workload more evenly or " | |
| f"add more resources ({constraint['match_count']} violations)" | |
| ) | |
| case _: | |
| suggestions.append( | |
| f"Constraint '{constraint['constraint_id']}' violated " | |
| f"({constraint['match_count']} times) - review constraint definition" | |
| ) | |
| if not suggestions: | |
| suggestions.append( | |
| "Hard constraints violated. Review constraint definitions and problem data." | |
| ) | |
| return suggestions | |