ContentsFoundations to FrontierBezhig

graduate_code/solutions/planning.py

All lab code. Download this file.

"""G10: deterministic critical path under unlimited-resource assumptions."""
import heapq
import math


def critical_path(durations, dependencies):
    if not 1 <= len(durations) <= 1000:
        raise ValueError("1..1000 tasks required")
    if any(type(name) is not str or not name for name in durations):
        raise ValueError("nonempty string task names required")
    if any(not math.isfinite(d) or d < 0 for d in durations.values()):
        raise ValueError("finite nonnegative durations required")
    if set(dependencies)-set(durations):
        raise ValueError("dependency map names an unknown task")
    before = {name:set(dependencies.get(name,())) for name in durations}
    if any(pred-set(durations) for pred in before.values()):
        raise ValueError("unknown predecessor")
    after = {name:[] for name in durations}
    for task,pred in before.items():
        for p in pred:
            after[p].append(task)
    degree = {name:len(pred) for name,pred in before.items()}
    ready = [name for name,d in degree.items() if d == 0]
    heapq.heapify(ready)
    order,earliest,finish = [],{},{}
    while ready:
        task = heapq.heappop(ready)
        order.append(task)
        earliest[task] = max((finish[p] for p in before[task]),default=0.)
        finish[task] = earliest[task]+durations[task]
        if not math.isfinite(finish[task]):
            raise ValueError("time overflow")
        for child in after[task]:
            degree[child] -= 1
            if degree[child] == 0:
                heapq.heappush(ready,child)
    if len(order) != len(durations):
        raise ValueError("dependency cycle")
    end = max(finish.values())
    latest = {}
    for task in reversed(order):
        end_task = min((latest[c] for c in after[task]),default=end)
        latest[task] = end_task-durations[task]
    return {"finish":end,"earliest":earliest,"latest":latest,
            "slack":{t:latest[t]-earliest[t] for t in durations},"order":order}


if __name__ == "__main__":
    print(critical_path({"A":3,"B":5,"C":4,"D":2},
                        {"B":["A"],"C":["A"],"D":["B","C"]}))