1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
# prediction_controller.py
from dateutil.relativedelta import relativedelta
from datetime import datetime, timedelta
from typing import List, Tuple, Dict, Optional
import math
from calendar_manager import CalendarManager
from date_calculator import DateCalculator
from event_type_handler import EventTypeHandler
from config import EventConfig
class PredictionController:
"""Handles prediction calculation with proper business logic orchestration"""
def __init__(self, calendar_manager: CalendarManager, date_calculator: DateCalculator):
self.calendar_manager = calendar_manager
self.date_calculator = date_calculator
self.launch_date: Optional[datetime] = None
self.duration: Optional[relativedelta] = None
self.prediction: Optional[datetime] = None
self.corrected_events: List[Tuple[datetime, datetime, str]] = []
def set_parameters(self, launch_date: str, duration_years: int) -> bool:
"""Set launch date and duration with validation"""
try:
self.launch_date = datetime.fromisoformat(launch_date)
self.duration = relativedelta(years=duration_years)
# Validate parameters
if duration_years <= 0:
raise ValueError("Duration must be positive")
if self.launch_date < datetime(1900, 1, 1):
raise ValueError("Launch date too far in the past")
if self.launch_date > datetime(2100, 1, 1):
raise ValueError("Launch date too far in the future")
return True
except ValueError as e:
print(f"Error setting parameters: {e}")
return False
def make_prediction(self, launch_date: str, duration_years: int) -> bool:
"""Calculate prediction with proper business logic orchestration"""
if not self.set_parameters(launch_date, duration_years):
return False
try:
# Calculate base prediction (launch + duration - 1 day)
prediction_start = self.launch_date + self.duration - timedelta(days=1)
# Categorize events by type
categorized_events = EventTypeHandler.categorize_events(self.calendar_manager.entries)
# Process events according to business rules
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.calendar_manager.correct_dates(all_corrected_events)
return True
except Exception as e:
print(f"Error calculating prediction: {e}")
return False
def get_launch_date(self) -> Optional[datetime]:
"""Get the launch date"""
return self.launch_date
def get_duration(self) -> Optional[relativedelta]:
"""Get the duration"""
return self.duration
def get_prediction(self) -> Optional[datetime]:
"""Get the prediction"""
return self.prediction
def validate_prediction_inputs(self, launch_date: str, duration_years: int) -> List[str]:
"""Validate prediction inputs and return list of errors"""
errors = []
try:
launch_dt = datetime.fromisoformat(launch_date)
except ValueError:
errors.append("Invalid launch date format")
return errors
if duration_years <= 0:
errors.append("Duration must be positive")
if launch_dt < datetime(1900, 1, 1):
errors.append("Launch date too far in the past")
if launch_dt > datetime(2100, 1, 1):
errors.append("Launch date too far in the future")
return errors
def save_complete_state(self, filename: str) -> bool:
"""Save the complete prediction state including all calculated data"""
from prediction_state_service import PredictionStateService
return PredictionStateService.save_prediction_state(
self.calendar_manager, self, filename
)
def load_complete_state(self, filename: str) -> bool:
"""Load complete prediction state and restore controller"""
from prediction_state_service import PredictionStateService
state = PredictionStateService.load_prediction_state(filename)
if state:
success = PredictionStateService.restore_from_state(
state, self.calendar_manager, self
)
if success and state.prediction_date:
# Restore prediction date directly if available
self.prediction = state.prediction_date
return success
return False
|
