Using a Directed Acyclic Graph algorithm to predict productivity

The world is empowered by productivity. Society, financial markets and economics are driven by the measure of it's productive output. Therefore, methods and frameworks designed to maximize output and efficiency by optimizing time management, task prioritization, and workflow are essential for both individuals and organizations striving for success.

In an era where automation and AI-driven tools are reshaping industries, the application of value-adding algorithms and processes are an important part the future of software.
In the following article I will demonstrate how using a Directed Acyclic Graph (DAG) can accurately predict task completion and project timelines.

A Directed Acyclic Graph (DAG) represents a structured workflow where tasks (nodes) are connected by dependencies (edges) in a clear, one-way sequence, ensuring no circular dependencies or infinite loops. This structure is crucial in project planning where certain tasks must be completed before others can begin. For example, in software development, "Design UI" must precede "Implement UI," which then must precede "Testing." Since DAGs maintain a topological order, they enable efficient scheduling, parallel execution of independent tasks, and prevention of deadlocks, ensuring smooth and optimized project execution.

Using the following example dataset:

tasks = [
    { 
        "id": 1, 
        "name": "Design UI", 
        "days_length": 3, 
        "dependencies": [] 
    },
    { 
        "id": 2, 
        "name": "Implement UI", 
        "days_length": 7, 
        "dependencies": [1] 
    },
    { 
        "id": 3, 
        "name": "Testing", 
        "days_length": 2, 
        "dependencies": [2] 
    },
    { 
        "id": 4, 
        "name": "Project Documentation", 
        "days_length": 4, 
        "dependencies": [] 
    }
]

To determine the correct order of tasks, we need to perform a topological sort of the tasks which will ensure that dependent tasks are performed before their edges. The first step of the process is to build an adjacency list containing each task and their in-degrees (tasks dependent on it). Once we have this we can begin ordering the schedule of tasks by adding in all tasks without any incoming edges. Using our sample data, this will be Project Documentation and Design UI. Once we have added these to our list, we can iterate over each of those tasks and then add each of their dependencies to our schedule.

from collections import defaultdict, deque
from datetime import datetime, timedelta, time

def topological_sort(self, tasks):
    # Step 1: Calculate in-degrees and construct the adjacency list
    in_degree = defaultdict(int)
    adjacency_list = defaultdict(list) 
    for task in tasks.values():
        # Initialize in-degree for the current task
        in_degree[task.id] = in_degree.get(task.id, 0)
        for dependency in task.dependencies:
            adjacency_list[dependency.id].append(task)
            in_degree[task.id] += 1

    # Step 2: Collect all tasks with in-degree 0 (tasks with no dependencies) for initial adding
    zero_in_degree = deque([task for task in tasks.values() if in_degree[task.id] == 0])

    # Step 3: Process tasks with in-degree 0
    top_order = []
    while zero_in_degree:
        # 3.1: add tasks that have no dependencies
        current_task = zero_in_degree.popleft()
        top_order.append(current_task)

        # Reduce the in-degree of dependent tasks
        # 3.2: process dependent tasks 
        # get list of tasks that point to this task
        for dependent_task in adjacency_list[current_task.id]:
            # subtract -1 to account for the parent task that points to this
            in_degree[dependent_task.id] -= 1  # Step 1: Reduce the in-degree of each dependent task

            # this dependent task has no other dependencies
            if in_degree[dependent_task.id] == 0: # Step 2: Check if this dependent task now has no incoming edges (all depedendencies are resolved)
                zero_in_degree.append(tasks[dependent_task.id]) # Step 3: Add it to the processing queue (tasks with no incoming edges)

    return top_order

While the code example above will order all tasks in their required order, by adding logic to determine start and end dates for each tasks based on anticipated task length, we can extrapolate the anticipate completion date of each task and accurately predict project timeframe and completion.