feat(algorithms, intervals): task scheduler problem

Author: BrianLusina

algorithms/intervals/__init__.py

@@ -0,0 +1,20 @@
+# Task Scheduler
+
+You are given an array of CPU tasks represented by uppercase letters (A to Z) and an integer n, which denotes the 
+cooling period required between any two identical tasks. Each task takes exactly one CPU interval to execute. Therefore,
+each CPU interval can either perform a task or remain idle. Tasks can be executed in any order, but the same task must 
+be separated by at least n intervals.
+
+Determine the minimum number of CPU intervals required to complete all tasks.
+
+**Constraints**
+
+- 1 ≤ `tasks.length` ≤ 1000
+- 0 ≤ `n` ≤ 100
+- `tasks` consists of uppercase English letters
+
+## Examples
+
+![Example 1](./images/examples/task_scheduler_example_1.png)
+![Example 2](./images/examples/task_scheduler_example_2.png)
+![Example 3](./images/examples/task_scheduler_example_3.png)

algorithms/intervals/task_scheduler/README.md

@@ -0,0 +1,97 @@
+from typing import List
+from collections import Counter
+from heapq import heapify, heappop, heappush
+
+
+def least_intervals_mathematical(tasks: List[str], n: int) -> int:
+    """
+    The time complexity of is O(N), where N is the number of tasks:
+    Counter(tasks) takes O(N) time.
+    The rest of the calculation is constant time, O(1), because the number of unique tasks (A-Z) is fixed at 26, meaning
+     task_count.values() is at most 26 elements long.
+
+     This greedy approach, which relies on the mathematical structure imposed by the most frequent task, is the most
+     efficient way to solve the Task Scheduler problem.
+    """
+    # get the total number of tasks
+    total_tasks = len(tasks)
+
+    # Step 1: Use Counter to get all frequencies
+    task_count = Counter(tasks)
+    if not task_count:
+        return 0
+
+    # most_common() returns a list of (task, frequency) tuples
+    # most_common(1) gives the highest frequency task: [('A', 3)] -> max_freq = 3
+    max_freq = task_count.most_common(1)[0][1]
+
+    # Step 2: Calculate how many tasks share the max frequency
+    # This is slightly more efficient than looping through the full dictionary
+    # because it stops checking once frequencies drop below max_freq.
+    max_freq_count = sum(1 for freq in task_count.values() if freq == max_freq)
+
+    # Step 3: Calculate the least number of intervals
+    result = (max_freq - 1) * (n + 1) + max_freq_count
+
+    # Step 4: Return the maximum of result and total tasks
+    return max(result, total_tasks)
+
+
+def least_intervals_with_max_heap(tasks: List[str], n: int) -> int:
+    """
+    Calculates the minimum CPU intervals required using an iterative Max Heap approach.
+
+    This Max Heap solution is slightly less performant than the mathematical one, but it explicitly models the idle time
+    (slots where max_heap is empty mid-cycle).
+
+    Max Heap Complexity: O(T⋅KlogK), where T is the total number of intervals (can be up to N⋅n), and K is the number
+    of unique tasks (at most 26).
+    """
+    # 1. Count Frequencies
+    task_count = Counter(tasks)
+    if not task_count:
+        return 0
+
+    time = 0
+    # 2. Setup Max Heap (store negative frequencies for a Python Min Heap)
+    # The task with the highest frequency will have the smallest negative value (e.g., -3).
+    max_heap = [-freq for freq in task_count.values()]
+    heapify(max_heap)
+
+    # Loop continues until all tasks are executed (heap is empty)
+    while max_heap:
+        temp_list = []
+        cycle_len = 0
+
+        # Execute tasks for one cooling cycle (n + 1 slots)
+        # We process up to n + 1 tasks, always picking the highest frequency one available.
+        for _ in range(n + 1):
+            if max_heap:
+                # Pop the highest frequency task
+                frequency = -heappop(max_heap)
+
+                if frequency > 1:
+                    # If the task needs to run again, store the remaining count
+                    temp_list.append(frequency - 1)
+
+                cycle_len += 1
+            else:
+                # Stop iterating slots if the heap is empty, though time will still advance.
+                break
+
+        # 3. Re-insert Tasks
+        # Push the remaining counts back onto the heap (as negative values)
+        for freq in temp_list:
+            heappush(max_heap, -freq)
+
+        # 4. Time Calculation
+        if max_heap:
+            # If the heap is NOT empty, it means there are still tasks remaining.
+            # We must wait for the full cooling period, so we advance time by n + 1.
+            time += n + 1
+        else:
+            # If the heap IS empty, it means all tasks were completed in this cycle.
+            # We only count the intervals actually used (cycle_len).
+            time += cycle_len
+
+    return time

algorithms/intervals/task_scheduler/__init__.py

@@ -0,0 +1,135 @@
+import unittest
+from typing import List
+from parameterized import parameterized
+from algorithms.intervals.task_scheduler import (
+    least_intervals_mathematical,
+    least_intervals_with_max_heap,
+)
+
+
+class LeastIntervalsTestCase(unittest.TestCase):
+    @parameterized.expand(
+        [
+            (["A", "A", "B", "B"], 2, 5),
+            (["A", "B", "A", "A", "B", "C"], 3, 9),
+            (["A", "C", "B", "A"], 0, 4),
+            (["A", "A", "A", "B", "B", "C", "C"], 1, 7),
+            (["S", "I", "V", "U", "W", "D", "U", "X"], 0, 8),
+            (
+                [
+                    "A",
+                    "K",
+                    "X",
+                    "M",
+                    "W",
+                    "D",
+                    "X",
+                    "B",
+                    "D",
+                    "C",
+                    "O",
+                    "Z",
+                    "D",
+                    "E",
+                    "Q",
+                ],
+                3,
+                15,
+            ),
+            (
+                [
+                    "A",
+                    "B",
+                    "C",
+                    "O",
+                    "Q",
+                    "C",
+                    "Z",
+                    "O",
+                    "X",
+                    "C",
+                    "W",
+                    "Q",
+                    "Z",
+                    "B",
+                    "M",
+                    "N",
+                    "R",
+                    "L",
+                    "C",
+                    "J",
+                ],
+                10,
+                34,
+            ),
+        ]
+    )
+    def test_least_intervals_mathematical(
+        self, tasks: List[str], n: int, expected: int
+    ):
+        actual = least_intervals_mathematical(tasks, n)
+        self.assertEqual(expected, actual)
+
+    @parameterized.expand(
+        [
+            (["A", "A", "B", "B"], 2, 5),
+            (["A", "B", "A", "A", "B", "C"], 3, 9),
+            (["A", "C", "B", "A"], 0, 4),
+            (["A", "A", "A", "B", "B", "C", "C"], 1, 7),
+            (["S", "I", "V", "U", "W", "D", "U", "X"], 0, 8),
+            (
+                [
+                    "A",
+                    "K",
+                    "X",
+                    "M",
+                    "W",
+                    "D",
+                    "X",
+                    "B",
+                    "D",
+                    "C",
+                    "O",
+                    "Z",
+                    "D",
+                    "E",
+                    "Q",
+                ],
+                3,
+                15,
+            ),
+            (
+                [
+                    "A",
+                    "B",
+                    "C",
+                    "O",
+                    "Q",
+                    "C",
+                    "Z",
+                    "O",
+                    "X",
+                    "C",
+                    "W",
+                    "Q",
+                    "Z",
+                    "B",
+                    "M",
+                    "N",
+                    "R",
+                    "L",
+                    "C",
+                    "J",
+                ],
+                10,
+                34,
+            ),
+        ]
+    )
+    def test_least_intervals_max_heap(self, tasks: List[str], n: int, expected: int):
+        actual = least_intervals_with_max_heap(tasks, n)
+        self.assertEqual(expected, actual)
+
+
+if __name__ == "__main__":
+    unittest.main()