""" PM Module — Advanced Domain Services سرویس‌های پیشرفته: • ResourceLevelingService — تسطیح منابع (هیوریستیک) • MonteCarloService — تحلیل مونت‌کارلو • CCPMService — زنجیره بحرانی • ScheduleHealthService — ارزیابی سلامت زمان‌بندی """ from dataclasses import dataclass, field from datetime import date, timedelta from decimal import Decimal from typing import List, Optional, Dict, Tuple from uuid import UUID import random import math from ..entities.task import Task, Dependency from ..entities.resource import ResourceAssignment from ..entities.advanced import ScheduleHealthSnapshot from ..services.scheduling import CalendarData, DEFAULT_CALENDAR, _add_working_days # ═══════════════════════════════════════════════════ # Resource Leveling (تسطیح منابع — هیوریستیک ساده) # ═══════════════════════════════════════════════════ @dataclass(frozen=True) class LevelingResult: """نتیجه تسطیح منابع.""" leveled_tasks: Dict[UUID, date] # task_id → new start_date overallocated_resources: List[UUID] iterations: int success: bool message: str = "" class ResourceLevelingService: """ تسطیح منابع با الگوریتم هیوریستیک — Priority-Based Serial Leveling. ۱. تسک‌ها را بر اساس اولویت مرتب کن (Early Start, Total Float, Critical Path) ۲. برای هر تسک بررسی کن آیا منبع overallocate شده ۳. اگر بله، تسک را به جلو shift بده تا منبع آزاد شود ۴. بعد از هر shift، successorها را به‌روز‌رسانی کن """ MAX_ITERATIONS = 500 @staticmethod def level( tasks: List[Task], dependencies: List[Dependency], assignments: List[ResourceAssignment], resource_max_units: Dict[UUID, Decimal], calendar: CalendarData = DEFAULT_CALENDAR, ) -> LevelingResult: """اجرای تسطیح منابع.""" # ساخت lookup tables task_map: Dict[UUID, Task] = {t.id: t for t in tasks} successors_map: Dict[UUID, List[Dependency]] = {} for dep in dependencies: successors_map.setdefault(dep.predecessor_id, []).append(dep) # گروه‌بندی assignments بر اساس resource resource_assignments: Dict[UUID, List[ResourceAssignment]] = {} for a in assignments: resource_assignments.setdefault(a.resource_id, []).append(a) # ساخت task start dates (mutable copy) task_starts: Dict[UUID, date] = {} for t in tasks: task_starts[t.id] = t.planned_start or t.early_start or date.today() overallocated: set = set() iterations = 0 for iteration in range(ResourceLevelingService.MAX_ITERATIONS): iterations = iteration + 1 conflict_found = False # مرتب‌سازی تسک‌ها: اول Early Start، بعد Total Float (ascending = less float first) sorted_tasks = sorted( tasks, key=lambda t: ( task_starts.get(t.id, date.today()), t.total_float or 999, 0 if t.is_critical else 1, ), ) for task in sorted_tasks: if task.is_milestone or (task.duration or 0) == 0: continue task_start = task_starts[task.id] task_end = _add_working_days( task_start, task.duration or 0, calendar ) # بررسی هر منبع assign شده به این تسک task_assignments = [ a for a in assignments if a.task_id == task.id and a.is_active ] for assignment in task_assignments: rid = assignment.resource_id max_units = resource_max_units.get(rid, Decimal("100")) # محاسبه بار منبع در بازه تسک total_units = Decimal("0") for other_a in resource_assignments.get(rid, []): if other_a.task_id == task.id or not other_a.is_active: continue other_task = task_map.get(other_a.task_id) if not other_task: continue other_start = task_starts.get(other_task.id, date.today()) other_end = _add_working_days( other_start, other_task.duration or 0, calendar ) # بررسی overlap if other_start < task_end and other_end > task_start: total_units += other_a.units or Decimal("0") total_units += assignment.units or Decimal("0") if total_units > max_units: overallocated.add(rid) conflict_found = True # Shift تسک به جلو: ۱ روز کاری new_start = _add_working_days(task_start, 1, calendar) task_starts[task.id] = new_start # Cascade to successors ResourceLevelingService._cascade_successors( task.id, new_start, task.duration or 0, successors_map, task_starts, task_map, calendar, ) break if not conflict_found: break return LevelingResult( leveled_tasks=task_starts, overallocated_resources=list(overallocated), iterations=iterations, success=not bool(overallocated) or iterations < ResourceLevelingService.MAX_ITERATIONS, message="تسطیح با موفقیت انجام شد" if iterations < ResourceLevelingService.MAX_ITERATIONS else "حداکثر تکرار به پایان رسید — ممکن است conflict باقی مانده باشد", ) @staticmethod def _cascade_successors( task_id: UUID, new_start: date, duration: int, successors_map: Dict, task_starts: Dict, task_map: Dict, calendar: CalendarData, ): """بروزرسانی زنجیره‌ای successorها پس از shift.""" new_end = _add_working_days(new_start, duration, calendar) for dep in successors_map.get(task_id, []): succ = task_map.get(dep.successor_id) if not succ: continue lag = dep.lag_days or 0 required_start = _add_working_days(new_end, lag, calendar) current_start = task_starts.get(succ.id, date.today()) if required_start > current_start: task_starts[succ.id] = required_start ResourceLevelingService._cascade_successors( succ.id, required_start, succ.duration or 0, successors_map, task_starts, task_map, calendar, ) # ═══════════════════════════════════════════════════ # Monte Carlo Simulation (شبیه‌سازی مونت‌کارلو) # ═══════════════════════════════════════════════════ @dataclass(frozen=True) class MonteCarloResult: """نتیجه شبیه‌سازی مونت‌کارلو.""" iterations: int mean_duration: float std_deviation: float p50_duration: float # median p80_duration: float p90_duration: float p95_duration: float min_duration: float max_duration: float histogram: List[Tuple[float, int]] # (bin_edge, count) confidence_range: Tuple[float, float] # (low, high) for 80% confidence @dataclass class TaskEstimate: """تخمین سه‌گانه (PERT) برای تسک.""" task_id: UUID optimistic: float # خوش‌بینانه most_likely: float # محتمل‌ترین pessimistic: float # بدبینانه class MonteCarloService: """ تحلیل مونت‌کارلو — محاسبه توزیع مدت پروژه. برای هر iteration: ۱. مدت هر تسک را با PERT/Triangular distribution نمونه‌گیری کن ۲. CPM را با مدت نمونه‌گیری‌شده اجرا کن ۳. مدت پروژه را ثبت کن نتیجه: توزیع آماری مدت پروژه """ DEFAULT_ITERATIONS = 1000 @staticmethod def simulate( tasks: List[Task], dependencies: List[Dependency], estimates: List[TaskEstimate], iterations: int = DEFAULT_ITERATIONS, ) -> MonteCarloResult: """اجرای شبیه‌سازی مونت‌کارلو.""" estimate_map: Dict[UUID, TaskEstimate] = {e.task_id: e for e in estimates} task_map: Dict[UUID, Task] = {t.id: t for t in tasks} # ساخت dependency graph predecessors_map: Dict[UUID, List[Dependency]] = {} for dep in dependencies: predecessors_map.setdefault(dep.successor_id, []).append(dep) # تعیین ترتیب توپولوژیکی topo_order = MonteCarloService._topological_sort(tasks, dependencies) results: List[float] = [] for _ in range(iterations): # نمونه‌گیری مدت هر تسک sampled_durations: Dict[UUID, float] = {} for task in tasks: est = estimate_map.get(task.id) if est: sampled = MonteCarloService._sample_pert( est.optimistic, est.most_likely, est.pessimistic ) else: sampled = float(task.duration or 0) sampled_durations[task.id] = max(0, sampled) # Forward pass (ساده — بدون تقویم) early_finish: Dict[UUID, float] = {} early_start: Dict[UUID, float] = {} for task_id in topo_order: es = 0.0 for dep in predecessors_map.get(task_id, []): pred_ef = early_finish.get(dep.predecessor_id, 0.0) lag = float(dep.lag_days or 0) es = max(es, pred_ef + lag) early_start[task_id] = es early_finish[task_id] = es + sampled_durations.get(task_id, 0.0) # مدت پروژه = max early_finish project_duration = max(early_finish.values()) if early_finish else 0.0 results.append(project_duration) # تحلیل آماری results.sort() n = len(results) mean_val = sum(results) / n if n else 0 variance = sum((x - mean_val) ** 2 for x in results) / n if n else 0 std_dev = math.sqrt(variance) p50 = results[int(n * 0.50)] if n else 0 p80 = results[int(n * 0.80)] if n else 0 p90 = results[int(n * 0.90)] if n else 0 p95 = results[min(int(n * 0.95), n - 1)] if n else 0 # هیستوگرام (10 بین) min_val = results[0] if results else 0 max_val = results[-1] if results else 0 bin_count = 10 bin_width = (max_val - min_val) / bin_count if max_val != min_val else 1 histogram: List[Tuple[float, int]] = [] for i in range(bin_count): edge = min_val + i * bin_width count = sum(1 for x in results if edge <= x < edge + bin_width) histogram.append((round(edge, 1), count)) # 80% confidence range low_idx = int(n * 0.10) high_idx = int(n * 0.90) return MonteCarloResult( iterations=iterations, mean_duration=round(mean_val, 1), std_deviation=round(std_dev, 1), p50_duration=round(p50, 1), p80_duration=round(p80, 1), p90_duration=round(p90, 1), p95_duration=round(p95, 1), min_duration=round(min_val, 1), max_duration=round(max_val, 1), histogram=histogram, confidence_range=( round(results[low_idx], 1) if n else 0, round(results[high_idx], 1) if n else 0, ), ) @staticmethod def _sample_pert(optimistic: float, most_likely: float, pessimistic: float) -> float: """نمونه‌گیری با توزیع PERT (Beta-PERT).""" if optimistic >= pessimistic: return most_likely # PERT mean mu = (optimistic + 4 * most_likely + pessimistic) / 6 # Use triangular distribution as approximation return random.triangular(optimistic, pessimistic, most_likely) @staticmethod def _topological_sort(tasks: List[Task], dependencies: List[Dependency]) -> List[UUID]: """مرتب‌سازی توپولوژیکی تسک‌ها.""" in_degree: Dict[UUID, int] = {t.id: 0 for t in tasks} adj: Dict[UUID, List[UUID]] = {t.id: [] for t in tasks} for dep in dependencies: if dep.successor_id in in_degree and dep.predecessor_id in adj: adj[dep.predecessor_id].append(dep.successor_id) in_degree[dep.successor_id] += 1 queue = [tid for tid, deg in in_degree.items() if deg == 0] result = [] while queue: node = queue.pop(0) result.append(node) for succ in adj.get(node, []): in_degree[succ] -= 1 if in_degree[succ] == 0: queue.append(succ) # اگر cycle وجود داشته باشد، بقیه تسک‌ها را هم اضافه کن remaining = [t.id for t in tasks if t.id not in set(result)] return result + remaining # ═══════════════════════════════════════════════════ # Critical Chain Project Management (CCPM) # ═══════════════════════════════════════════════════ @dataclass(frozen=True) class BufferInfo: """اطلاعات بافر.""" buffer_type: str # project | feeding size_days: int consumed_days: int = 0 consumption_percent: float = 0.0 status: str = "green" # green | yellow | red @property def remaining_days(self) -> int: return max(0, self.size_days - self.consumed_days) @dataclass(frozen=True) class CCPMResult: """نتیجه محاسبه CCPM.""" critical_chain: List[UUID] # task IDs in critical chain order project_buffer: BufferInfo feeding_buffers: Dict[UUID, BufferInfo] # merging_task_id → buffer total_project_duration: int # with buffers aggressive_duration: int # without buffers class CCPMService: """ مدیریت زنجیره بحرانی (Critical Chain Project Management). ۱. Critical Chain را شناسایی کن (با در نظر گرفتن منابع) ۲. Safety margin تسک‌ها را حذف کن (50% of safety) ۳. Project Buffer = 50% of critical chain duration (cut) ۴. Feeding Buffers = 50% of non-critical paths feeding into chain """ BUFFER_CUT_RATIO = 0.5 # نسبت برش ایمنی @staticmethod def calculate( tasks: List[Task], dependencies: List[Dependency], critical_path_ids: List[UUID], ) -> CCPMResult: """محاسبه CCPM — بافرها و زنجیره بحرانی.""" task_map: Dict[UUID, Task] = {t.id: t for t in tasks} critical_set = set(critical_path_ids) # ═══ ۱. محاسبه مدت تهاجمی (aggressive) ═══ # فرض: safety = 50% of duration, aggressive = duration * 0.5 aggressive_durations: Dict[UUID, int] = {} total_safety_cut = 0 for task in tasks: d = task.duration or 0 if d > 0: safety = int(d * CCPMService.BUFFER_CUT_RATIO) aggressive_durations[task.id] = d - safety if task.id in critical_set: total_safety_cut += safety else: aggressive_durations[task.id] = 0 # ═══ ۲. Project Buffer ═══ # SSQ method: sqrt(sum of squares of safety cuts) critical_safety_cuts = [] for tid in critical_path_ids: d = task_map[tid].duration or 0 if d > 0: critical_safety_cuts.append(int(d * CCPMService.BUFFER_CUT_RATIO)) project_buffer_size = int( math.sqrt(sum(c ** 2 for c in critical_safety_cuts)) ) if critical_safety_cuts else 0 # ═══ ۳. Feeding Buffers ═══ # پیدا کردن نقاط merge (تسک‌های غیربحرانی که به زنجیره بحرانی وارد می‌شوند) predecessors_map: Dict[UUID, List[Dependency]] = {} for dep in dependencies: predecessors_map.setdefault(dep.successor_id, []).append(dep) feeding_buffers: Dict[UUID, BufferInfo] = {} for crit_id in critical_path_ids: for dep in predecessors_map.get(crit_id, []): pred_id = dep.predecessor_id if pred_id not in critical_set: # Feeding path found feeding_chain = CCPMService._trace_feeding_chain( pred_id, critical_set, predecessors_map, task_map ) feeding_safety_cuts = [] for fid in feeding_chain: d = task_map[fid].duration or 0 if d > 0: feeding_safety_cuts.append( int(d * CCPMService.BUFFER_CUT_RATIO) ) fb_size = int( math.sqrt(sum(c ** 2 for c in feeding_safety_cuts)) ) if feeding_safety_cuts else 0 if fb_size > 0: feeding_buffers[crit_id] = BufferInfo( buffer_type="feeding", size_days=fb_size, ) # ═══ ۴. مجموع مدت ═══ aggressive_total = sum( aggressive_durations.get(tid, 0) for tid in critical_path_ids ) total_with_buffers = aggressive_total + project_buffer_size return CCPMResult( critical_chain=critical_path_ids, project_buffer=BufferInfo( buffer_type="project", size_days=project_buffer_size, ), feeding_buffers=feeding_buffers, total_project_duration=total_with_buffers, aggressive_duration=aggressive_total, ) @staticmethod def calculate_buffer_consumption( buffer: BufferInfo, completion_percent: float ) -> BufferInfo: """محاسبه مصرف بافر و وضعیت ترافیک لایت.""" if buffer.size_days == 0: return buffer consumption_pct = buffer.consumed_days / buffer.size_days * 100 expected_consumption_pct = completion_percent # Fever Chart thresholds if consumption_pct <= expected_consumption_pct * 0.33: status = "green" elif consumption_pct <= expected_consumption_pct * 0.67: status = "yellow" else: status = "red" return BufferInfo( buffer_type=buffer.buffer_type, size_days=buffer.size_days, consumed_days=buffer.consumed_days, consumption_percent=round(consumption_pct, 1), status=status, ) @staticmethod def _trace_feeding_chain( task_id: UUID, critical_set: set, predecessors_map: Dict, task_map: Dict, ) -> List[UUID]: """ردیابی زنجیره تغذیه (feeding chain) از یک تسک غیربحرانی.""" chain = [task_id] current = task_id visited = {task_id} while True: preds = predecessors_map.get(current, []) non_critical_preds = [ d.predecessor_id for d in preds if d.predecessor_id not in critical_set and d.predecessor_id not in visited and d.predecessor_id in task_map ] if not non_critical_preds: break # انتخاب longest predecessor best = max( non_critical_preds, key=lambda pid: task_map[pid].duration or 0, ) chain.append(best) visited.add(best) current = best return chain # ═══════════════════════════════════════════════════ # Schedule Health Assessment (ارزیابی سلامت زمان‌بندی) # ═══════════════════════════════════════════════════ class ScheduleHealthService: """ ارزیابی سلامت زمان‌بندی — الهام از DCMA 14-Point Assessment. ۱. Logic (Missing predecessors/successors) ۲. Leads (negative lag) ۳. Lags (excessive positive lag) ۴. Relationship Types (FS preferred) ۵. Hard Constraints ۶. High Float (>44 days) ۷. Negative Float ۸. High Duration (>44 days) ۹. Invalid Dates ۱۰. Resources Assigned ۱۱. Missed Tasks ۱۲. Critical Path % ۱۳. BEI (Baseline Execution Index) ۱۴. CPLI (Critical Path Length Index) """ HIGH_FLOAT_THRESHOLD = 44 # روز HIGH_DURATION_THRESHOLD = 44 # روز @staticmethod def assess( tasks: List[Task], dependencies: List[Dependency], assignments: List[ResourceAssignment], baseline_end: Optional[date] = None, data_date: Optional[date] = None, ) -> ScheduleHealthSnapshot: """ارزیابی کامل سلامت زمان‌بندی.""" if not data_date: data_date = date.today() total = len(tasks) if total == 0: return ScheduleHealthSnapshot(health_score=100) # ساخت lookups task_ids = {t.id for t in tasks} pred_map: Dict[UUID, List[UUID]] = {} succ_map: Dict[UUID, List[UUID]] = {} for dep in dependencies: pred_map.setdefault(dep.successor_id, []).append(dep.predecessor_id) succ_map.setdefault(dep.predecessor_id, []).append(dep.successor_id) assigned_tasks = {a.task_id for a in assignments if a.is_active} details = {} score = 100 # شروع از 100 و کسر کردن # ═══ 1. Missing Logic ═══ missing_logic = 0 for t in tasks: if t.is_milestone: continue has_pred = t.id in pred_map has_succ = t.id in succ_map if not has_pred and not has_succ: missing_logic += 1 missing_logic_pct = missing_logic / total * 100 details["missing_logic"] = { "count": missing_logic, "percent": round(missing_logic_pct, 1), "threshold": 5, "status": "pass" if missing_logic_pct <= 5 else "fail", } if missing_logic_pct > 5: score -= min(15, int(missing_logic_pct)) # ═══ 2-3. Leads & Lags ═══ leads_count = sum(1 for d in dependencies if (d.lag_days or 0) < 0) lags_count = sum(1 for d in dependencies if (d.lag_days or 0) > 5) total_deps = len(dependencies) or 1 leads_pct = leads_count / total_deps * 100 lags_pct = lags_count / total_deps * 100 details["leads"] = {"count": leads_count, "percent": round(leads_pct, 1)} details["lags"] = {"count": lags_count, "percent": round(lags_pct, 1)} if leads_pct > 5: score -= min(5, int(leads_pct)) # ═══ 4. Relationship Types ═══ fs_count = sum(1 for d in dependencies if d.dependency_type == "FS") fs_pct = fs_count / total_deps * 100 if total_deps > 1 else 100 details["fs_relationships"] = {"percent": round(fs_pct, 1)} if fs_pct < 90: score -= min(5, int((90 - fs_pct) / 5)) # ═══ 5. Hard Constraints ═══ hard_constraint_count = sum( 1 for t in tasks if hasattr(t, "constraint_type") and t.constraint_type and t.constraint_type not in ("asap", "alap") ) hard_pct = hard_constraint_count / total * 100 details["hard_constraints"] = { "count": hard_constraint_count, "percent": round(hard_pct, 1) } if hard_pct > 5: score -= min(5, int(hard_pct)) # ═══ 6. High Float ═══ high_float = sum( 1 for t in tasks if (t.total_float or 0) > ScheduleHealthService.HIGH_FLOAT_THRESHOLD ) high_float_pct = high_float / total * 100 details["high_float"] = { "count": high_float, "percent": round(high_float_pct, 1) } if high_float_pct > 5: score -= min(10, int(high_float_pct)) # ═══ 7. Negative Float ═══ negative_float = sum(1 for t in tasks if (t.total_float or 0) < 0) neg_float_pct = negative_float / total * 100 details["negative_float"] = { "count": negative_float, "percent": round(neg_float_pct, 1) } if negative_float > 0: score -= min(15, negative_float * 3) # ═══ 8. High Duration ═══ high_duration = sum( 1 for t in tasks if (t.duration or 0) > ScheduleHealthService.HIGH_DURATION_THRESHOLD and not t.is_milestone ) details["high_duration"] = {"count": high_duration} if high_duration > 0: score -= min(5, high_duration) # ═══ 9. Invalid Dates ═══ invalid_dates = sum( 1 for t in tasks if t.planned_start and t.planned_end and t.planned_start > t.planned_end ) details["invalid_dates"] = {"count": invalid_dates} if invalid_dates > 0: score -= min(10, invalid_dates * 5) # ═══ 10. Resources Assigned ═══ non_milestone = [t for t in tasks if not t.is_milestone] assigned_count = sum(1 for t in non_milestone if t.id in assigned_tasks) assigned_pct = assigned_count / len(non_milestone) * 100 if non_milestone else 100 details["resources_assigned"] = {"percent": round(assigned_pct, 1)} if assigned_pct < 90: score -= min(5, int((90 - assigned_pct) / 10)) # ═══ 11. Missed Tasks ═══ missed = sum( 1 for t in tasks if t.planned_end and t.planned_end < data_date and t.status not in ("completed", "cancelled") ) details["missed_tasks"] = {"count": missed} if missed > 0: score -= min(10, missed * 2) # ═══ 12. Critical Path Ratio ═══ critical_count = sum(1 for t in tasks if t.is_critical) critical_pct = critical_count / total * 100 details["critical_path"] = { "count": critical_count, "percent": round(critical_pct, 1) } # ═══ 13-14. BEI & CPLI ═══ bei = None cpli = None if baseline_end and critical_count > 0: # BEI = (Completed tasks / Total tasks) / (Elapsed time / Total planned time) completed_count = sum(1 for t in tasks if t.status == "completed") project_start = min( (t.planned_start for t in tasks if t.planned_start), default=data_date ) elapsed = (data_date - project_start).days or 1 total_planned = (baseline_end - project_start).days or 1 if total_planned > 0 and elapsed > 0: bei = round( Decimal(str((completed_count / total) / (elapsed / total_planned))), 3, ) # CPLI = (Remaining critical path + Project Buffer) / (Baseline End - Data Date) remaining_cp_days = sum( (t.duration or 0) for t in tasks if t.is_critical and t.status != "completed" ) remaining_calendar = (baseline_end - data_date).days if remaining_calendar > 0: cpli = round( Decimal(str(remaining_cp_days / remaining_calendar)), 3, ) out_of_sequence = 0 # placeholder — نیاز به بررسی actual vs planned health_score = max(0, min(100, score)) return ScheduleHealthSnapshot( snapshot_date=data_date, bei=bei, cpli=cpli, critical_tasks_count=critical_count, total_tasks_count=total, out_of_sequence_count=out_of_sequence, missing_logic_count=missing_logic, negative_float_count=negative_float, high_float_count=high_float, health_score=health_score, details=details, )