diff options
Diffstat (limited to 'prediction_controller.py')
| -rw-r--r-- | prediction_controller.py | 200 |
1 files changed, 52 insertions, 148 deletions
diff --git a/prediction_controller.py b/prediction_controller.py index e822f2e..bb30ef7 100644 --- a/prediction_controller.py +++ b/prediction_controller.py @@ -52,163 +52,67 @@ class PredictionController: categorized_events = EventTypeHandler.categorize_events(self.calendar_manager.entries) # Process events according to business rules - processed_events = self._process_events_by_type(categorized_events) - - # Calculate final prediction - self.prediction = self._calculate_final_prediction(prediction_start, processed_events) + full_projects = [] + half_projects = [] + events = [] + + for event_type in ["EZ 100%", "EZ pauschal"]: + if event_type in categorized_events: + full_projects.extend(categorized_events[event_type]) + + if "EZ 50%" in categorized_events: + half_projects.extend(categorized_events["EZ 50%"]) + + if "Sonstige" in categorized_events: + events.extend(categorized_events["Sonstige"]) + + sorted_projects = DateCalculator.sort_periods(full_projects) + sorted_half_projects = DateCalculator.sort_periods(half_projects) + sorted_events = DateCalculator.sort_periods(events) + + considered_events = DateCalculator.truncate_periods(sorted_events, self.launch_date) + considered_full_projects = DateCalculator.truncate_periods(sorted_projects, self.launch_date) + considered_half_projects = DateCalculator.truncate_periods(sorted_half_projects, self.launch_date) + + considered_full_projects_rounded, months = DateCalculator.round_periods(considered_full_projects) + + non_overlapping_half_projects = [] + for test_interval in considered_half_projects: + non_overlapping_half_projects.extend( + DateCalculator.find_non_overlapping_periods(considered_full_projects_rounded, test_interval) + ) + + considered_half_projects_rounded, months2 = DateCalculator.round_periods(non_overlapping_half_projects) + + all_projects_merged = DateCalculator.sort_periods(considered_full_projects_rounded + considered_half_projects_rounded) + merged_event_periods = DateCalculator.adjust_periods(considered_events) + + non_overlapping_event_periods = [] + for test_interval in merged_event_periods: + non_overlapping_event_periods.extend( + DateCalculator.find_non_overlapping_periods(all_projects_merged, test_interval) + ) + + total_months = months + math.ceil(months2 / 2) + total_days = sum((end - start).days + 1 for start, end, _ in non_overlapping_event_periods) + prediction = self.launch_date + self.duration + relativedelta(months=total_months) + timedelta(days=total_days-1) + + # Calculate final prediction + max_prediction = prediction_start + EventConfig.get_max_prediction_duration() + self.prediction = min(prediction, max_prediction) + + # Collect corrected events from all categories + all_corrected_events = considered_full_projects_rounded + considered_half_projects_rounded + non_overlapping_event_periods # Apply corrections to calendar entries - self._apply_corrections_to_calendar() + self.calendar_manager.correct_dates(all_corrected_events) return True except Exception as e: print(f"Error calculating prediction: {e}") return False - - def _process_events_by_type(self, categorized_events: Dict[str, List[Tuple[datetime, datetime, str]]]) -> Dict[str, List[Tuple[datetime, datetime, str]]]: - """Process events according to business rules""" - processed = {} - - # Process full-time projects first (EZ 100% and EZ pauschal) - full_projects = [] - for event_type in ["EZ 100%", "EZ pauschal"]: - if event_type in categorized_events: - full_projects.extend(categorized_events[event_type]) - - if full_projects: - processed["full_projects"] = self._process_full_projects(full_projects) - - # Process half-time projects (EZ 50%) - if "EZ 50%" in categorized_events: - processed["half_projects"] = self._process_half_projects( - categorized_events["EZ 50%"], - processed.get("full_projects", []) - ) - - # Process other events (Sonstige) - if "Sonstige" in categorized_events: - processed["other_events"] = self._process_other_events( - categorized_events["Sonstige"], - processed.get("full_projects", []), - processed.get("half_projects", []) - ) - - return processed - - def _process_full_projects(self, full_projects: List[Tuple[datetime, datetime, str]]) -> List[Tuple[datetime, datetime, str]]: - """Process full-time projects""" - # Sort and truncate periods - sorted_projects = DateCalculator.sort_periods(full_projects) - considered_projects = DateCalculator.truncate_periods(sorted_projects, self.launch_date) - - # Filter out any invalid periods - valid_projects = DateCalculator.filter_valid_periods(considered_projects) - - # Round to month boundaries - rounded_projects, total_months = DateCalculator.round_periods(valid_projects) - - return rounded_projects - - def _process_half_projects(self, half_projects: List[Tuple[datetime, datetime, str]], - full_projects: List[Tuple[datetime, datetime, str]]) -> List[Tuple[datetime, datetime, str]]: - """Process half-time projects""" - # Sort and truncate periods - sorted_projects = DateCalculator.sort_periods(half_projects) - considered_projects = DateCalculator.truncate_periods(sorted_projects, self.launch_date) - - # Filter out any invalid periods - valid_projects = DateCalculator.filter_valid_periods(considered_projects) - - # Find non-overlapping periods with full projects - non_overlapping_projects = [] - for test_interval in valid_projects: - non_overlapping_results = DateCalculator.find_non_overlapping_periods(full_projects, test_interval) - non_overlapping_projects.extend(non_overlapping_results) - - # Filter valid periods again after overlap processing - valid_non_overlapping = DateCalculator.filter_valid_periods(non_overlapping_projects) - - # Round to month boundaries - rounded_projects, total_months = DateCalculator.round_periods(valid_non_overlapping) - - return rounded_projects - - def _process_other_events(self, other_events: List[Tuple[datetime, datetime, str]], - full_projects: List[Tuple[datetime, datetime, str]], - half_projects: List[Tuple[datetime, datetime, str]]) -> List[Tuple[datetime, datetime, str]]: - """Process other events (Sonstige)""" - # Sort and truncate periods - sorted_events = DateCalculator.sort_periods(other_events) - considered_events = DateCalculator.truncate_periods(sorted_events, self.launch_date) - - # Filter out any invalid periods - valid_events = DateCalculator.filter_valid_periods(considered_events) - - # Find non-overlapping periods with all projects - all_projects = DateCalculator.sort_periods(full_projects + half_projects) - non_overlapping_events = [] - for test_interval in valid_events: - non_overlapping_results = DateCalculator.find_non_overlapping_periods(all_projects, test_interval) - non_overlapping_events.extend(non_overlapping_results) - - # Filter valid periods again after overlap processing - valid_non_overlapping = DateCalculator.filter_valid_periods(non_overlapping_events) - - # Adjust overlapping periods - adjusted_events = DateCalculator.adjust_periods(valid_non_overlapping) - - # Final validation to ensure all periods are valid - final_valid_events = DateCalculator.filter_valid_periods(adjusted_events) - return final_valid_events - - def _calculate_final_prediction(self, prediction_start: datetime, - processed_events: Dict[str, List[Tuple[datetime, datetime, str]]]) -> datetime: - """Calculate the final prediction date""" - # Calculate months from projects - total_months = 0 - - # Full projects count as full months - if "full_projects" in processed_events: - total_months += DateCalculator.calculate_total_months(processed_events["full_projects"]) - - # Half projects count as half months - if "half_projects" in processed_events: - half_months = DateCalculator.calculate_total_months(processed_events["half_projects"]) - total_months += math.ceil(half_months / 2) - - # Other events count as days - total_days = 0 - if "other_events" in processed_events: - total_days = DateCalculator.calculate_total_days(processed_events["other_events"]) - - # Calculate final prediction - final_prediction = (prediction_start + - relativedelta(months=total_months) + - timedelta(days=total_days)) - - # Apply maximum limit - max_prediction = prediction_start + EventConfig.get_max_prediction_duration() - final_prediction = DateCalculator.min_date(final_prediction, max_prediction) - - return final_prediction - - def _apply_corrections_to_calendar(self): - """Apply corrected dates to calendar entries and calculate time_period""" - # Collect all corrected events from processed results - all_corrected_events = [] - - # Get processed events from the prediction calculation - categorized_events = EventTypeHandler.categorize_events(self.calendar_manager.entries) - processed_events = self._process_events_by_type(categorized_events) - - # Collect corrected events from all categories - for event_type, events in processed_events.items(): - all_corrected_events.extend(events) - - # Apply corrections to calendar entries - self.calendar_manager.correct_dates(all_corrected_events) def get_launch_date(self) -> Optional[datetime]: """Get the launch date""" |