10_gantt_chart.py

Download
python 186 lines 6.4 KB
  1"""
  2Gantt Chart Generator with Critical Path Method (CPM)
  3
  4Demonstrates project scheduling concepts from software engineering:
  5- Task definition with dependencies
  6- Forward pass (earliest start/finish)
  7- Backward pass (latest start/finish)
  8- Critical path identification
  9- Slack (float) calculation
 10- ASCII Gantt chart rendering
 11"""
 12
 13from dataclasses import dataclass, field
 14from datetime import date, timedelta
 15from typing import Optional
 16
 17
 18@dataclass
 19class Task:
 20    name: str
 21    duration: int          # working days
 22    dependencies: list[str] = field(default_factory=list)
 23    # Computed by CPM
 24    early_start: int = 0
 25    early_finish: int = 0
 26    late_start: int = 0
 27    late_finish: int = 0
 28
 29    @property
 30    def slack(self) -> int:
 31        return self.late_start - self.early_start
 32
 33    @property
 34    def is_critical(self) -> bool:
 35        return self.slack == 0
 36
 37
 38def forward_pass(tasks: dict[str, Task]) -> None:
 39    """Calculate earliest start and finish times."""
 40    resolved: set[str] = set()
 41
 42    def resolve(name: str) -> None:
 43        if name in resolved:
 44            return
 45        task = tasks[name]
 46        for dep in task.dependencies:
 47            resolve(dep)
 48        if task.dependencies:
 49            task.early_start = max(tasks[d].early_finish for d in task.dependencies)
 50        else:
 51            task.early_start = 0
 52        task.early_finish = task.early_start + task.duration
 53        resolved.add(name)
 54
 55    for name in tasks:
 56        resolve(name)
 57
 58
 59def backward_pass(tasks: dict[str, Task], project_duration: int) -> None:
 60    """Calculate latest start and finish times."""
 61    # Initialize: tasks with no successors finish at project end
 62    successors: dict[str, list[str]] = {name: [] for name in tasks}
 63    for name, task in tasks.items():
 64        for dep in task.dependencies:
 65            successors[dep].append(name)
 66
 67    resolved: set[str] = set()
 68
 69    def resolve(name: str) -> None:
 70        if name in resolved:
 71            return
 72        task = tasks[name]
 73        succs = successors[name]
 74        for s in succs:
 75            resolve(s)
 76        if succs:
 77            task.late_finish = min(tasks[s].late_start for s in succs)
 78        else:
 79            task.late_finish = project_duration
 80        task.late_start = task.late_finish - task.duration
 81        resolved.add(name)
 82
 83    # Resolve in reverse dependency order
 84    for name in reversed(list(tasks.keys())):
 85        resolve(name)
 86
 87
 88def find_critical_path(tasks: dict[str, Task]) -> list[str]:
 89    """Return task names on the critical path, in order."""
 90    critical = [name for name, t in tasks.items() if t.is_critical]
 91    # Sort by early_start to get execution order
 92    critical.sort(key=lambda n: tasks[n].early_start)
 93    return critical
 94
 95
 96def render_gantt(tasks: dict[str, Task], project_duration: int,
 97                 start_date: date, bar_width: int = 40) -> str:
 98    """Render an ASCII Gantt chart."""
 99    scale = bar_width / project_duration
100    lines: list[str] = []
101
102    # Header
103    name_col = max(len(t.name) for t in tasks.values()) + 2
104    lines.append(f"{'Task':<{name_col}} {'C':1} {'Slack':>5}  Timeline")
105    lines.append("-" * (name_col + 8 + bar_width))
106
107    # Date ruler (show week numbers)
108    ruler = [" "] * bar_width
109    for day in range(project_duration):
110        if day % 5 == 0:
111            label = str(day // 5 + 1)
112            pos = int(day * scale)
113            if pos < bar_width:
114                ruler[pos] = label
115    lines.append(f"{'':>{name_col}}   {'':>5}  {''.join(ruler)}")
116
117    for name, task in tasks.items():
118        bar = ["."] * bar_width
119        start_pos = int(task.early_start * scale)
120        end_pos = int(task.early_finish * scale)
121        end_pos = min(end_pos, bar_width)
122        for i in range(start_pos, end_pos):
123            bar[i] = "#" if task.is_critical else "="
124        critical_marker = "*" if task.is_critical else " "
125        bar_str = "".join(bar)
126        lines.append(
127            f"{name:<{name_col}} {critical_marker} {task.slack:>5}  {bar_str}"
128        )
129
130    lines.append("-" * (name_col + 8 + bar_width))
131    lines.append("Legend: # = critical path  = = non-critical  * = on critical path")
132
133    # Date range
134    end_date = start_date + timedelta(days=project_duration - 1)
135    lines.append(f"Project: {start_date}{end_date}  ({project_duration} working days)")
136    return "\n".join(lines)
137
138
139def schedule_project(tasks: dict[str, Task]) -> int:
140    """Run CPM and return total project duration."""
141    forward_pass(tasks)
142    project_duration = max(t.early_finish for t in tasks.values())
143    backward_pass(tasks, project_duration)
144    return project_duration
145
146
147if __name__ == "__main__":
148    # --- Example: E-Commerce Platform Development ---
149    tasks = {
150        "Requirements":    Task("Requirements",    5, []),
151        "UI Design":       Task("UI Design",       8, ["Requirements"]),
152        "DB Schema":       Task("DB Schema",       4, ["Requirements"]),
153        "Auth Service":    Task("Auth Service",    6, ["DB Schema"]),
154        "Product API":     Task("Product API",     7, ["DB Schema"]),
155        "Cart Service":    Task("Cart Service",    5, ["Auth Service", "Product API"]),
156        "Payment Integ":   Task("Payment Integ",  10, ["Auth Service"]),
157        "Frontend Dev":    Task("Frontend Dev",   12, ["UI Design", "Product API"]),
158        "Integration QA":  Task("Integration QA",  6, ["Cart Service", "Payment Integ",
159                                                        "Frontend Dev"]),
160        "UAT & Launch":    Task("UAT & Launch",    4, ["Integration QA"]),
161    }
162
163    project_duration = schedule_project(tasks)
164    critical_path = find_critical_path(tasks)
165    start_date = date(2025, 3, 3)  # Monday
166
167    print("=" * 65)
168    print("  E-COMMERCE PLATFORM — PROJECT SCHEDULE (CPM)")
169    print("=" * 65)
170
171    print(f"\nProject Duration : {project_duration} working days")
172    print(f"Critical Path    : {' → '.join(critical_path)}")
173    print(f"Start Date       : {start_date}")
174
175    print("\nTask Details:")
176    header = f"{'Task':<18} {'Dur':>4} {'ES':>4} {'EF':>4} {'LS':>4} {'LF':>4} {'Slack':>5}  Critical"
177    print(header)
178    print("-" * len(header))
179    for name, task in tasks.items():
180        marker = "YES" if task.is_critical else "—"
181        print(f"{name:<18} {task.duration:>4} {task.early_start:>4} "
182              f"{task.early_finish:>4} {task.late_start:>4} "
183              f"{task.late_finish:>4} {task.slack:>5}  {marker}")
184
185    print("\n" + render_gantt(tasks, project_duration, start_date))