Commit
·
f572466
1
Parent(s):
30f018f
refactor: add comprehensive type hints throughout codebase
Browse filesAdded type hints to all functions in app.py, schedule.py, and test_app.py
for better IDE support and static type checking. Includes complex generic
types for constraint solving functions and proper Gradio component typing.
🤖 Generated with [Claude Code](https://claude.ai/code)
Co-Authored-By: Claude <noreply@anthropic.com>
- app.py +25 -24
- schedule.py +1 -1
- test_app.py +5 -5
app.py
CHANGED
@@ -4,20 +4,21 @@ import re
|
|
4 |
from collections import defaultdict
|
5 |
from ortools.sat.python import cp_model
|
6 |
from datetime import datetime, timedelta
|
|
|
7 |
|
8 |
import pandas
|
9 |
|
10 |
|
11 |
-
def normalize_task_name(task):
|
12 |
"""Normalize task name to lowercase for consistent comparison."""
|
13 |
return task.strip().lower().replace(" ", "_")
|
14 |
|
15 |
|
16 |
-
def display_task_name(task):
|
17 |
return task
|
18 |
|
19 |
|
20 |
-
def parse_requirements(requirements):
|
21 |
"""Parse requirements list into dependency graph."""
|
22 |
dependencies = defaultdict(list)
|
23 |
all_tasks = set()
|
@@ -44,7 +45,7 @@ def parse_requirements(requirements):
|
|
44 |
return dependencies, list(all_tasks), original_names
|
45 |
|
46 |
|
47 |
-
def solve_all_tasks(dependencies, all_tasks):
|
48 |
"""Try to schedule all tasks (works if DAG)."""
|
49 |
model = cp_model.CpModel()
|
50 |
n_tasks = len(all_tasks)
|
@@ -71,7 +72,7 @@ def solve_all_tasks(dependencies, all_tasks):
|
|
71 |
return None
|
72 |
|
73 |
|
74 |
-
def solve_maximum_subset(dependencies, all_tasks):
|
75 |
"""Find maximum subset of tasks that can be executed (handles cycles)."""
|
76 |
model = cp_model.CpModel()
|
77 |
n_tasks = len(all_tasks)
|
@@ -119,7 +120,7 @@ def solve_maximum_subset(dependencies, all_tasks):
|
|
119 |
return []
|
120 |
|
121 |
|
122 |
-
def parse_tasks(tasks_text):
|
123 |
"""Parse tasks from text input."""
|
124 |
if not tasks_text.strip():
|
125 |
return [], {}
|
@@ -140,7 +141,7 @@ def parse_tasks(tasks_text):
|
|
140 |
return sorted(tasks), original_names
|
141 |
|
142 |
|
143 |
-
def generate_mermaid_gantt(task_order, original_names, durations=None, start_time=None):
|
144 |
"""Generate Mermaid Gantt chart syntax with minute granularity and selectable start time."""
|
145 |
if not task_order:
|
146 |
return "gantt\n title Task Execution Timeline\n dateFormat YYYY-MM-DD HH:mm\n section No Tasks\n No tasks to display : 2024-01-01 08:00, 1m"
|
@@ -177,7 +178,7 @@ def generate_mermaid_gantt(task_order, original_names, durations=None, start_tim
|
|
177 |
return gantt
|
178 |
|
179 |
|
180 |
-
def generate_mermaid_flowchart(dependencies, all_tasks, original_names):
|
181 |
"""Generate Mermaid flowchart syntax."""
|
182 |
if not all_tasks:
|
183 |
return "flowchart TD\n A[No tasks to display]"
|
@@ -206,7 +207,7 @@ def generate_mermaid_flowchart(dependencies, all_tasks, original_names):
|
|
206 |
return flowchart
|
207 |
|
208 |
|
209 |
-
def update_dropdowns(tasks_text):
|
210 |
"""Update dropdown choices when tasks change."""
|
211 |
tasks, original_names = parse_tasks(tasks_text)
|
212 |
|
@@ -224,7 +225,7 @@ def update_dropdowns(tasks_text):
|
|
224 |
)
|
225 |
|
226 |
|
227 |
-
def add_dependencies(task_display, required_display_list, current_deps):
|
228 |
"""Add multiple dependencies for a single task."""
|
229 |
if not task_display:
|
230 |
return current_deps, "⚠️ Please select a task"
|
@@ -281,7 +282,7 @@ def add_dependencies(task_display, required_display_list, current_deps):
|
|
281 |
return updated_deps, status
|
282 |
|
283 |
|
284 |
-
def remove_dependency(deps_to_remove, current_deps):
|
285 |
"""Remove selected dependencies."""
|
286 |
if not deps_to_remove:
|
287 |
return current_deps, "⚠️ Please select dependencies to remove"
|
@@ -291,7 +292,7 @@ def remove_dependency(deps_to_remove, current_deps):
|
|
291 |
return updated_deps, f"✅ Removed {removed_count} dependencies"
|
292 |
|
293 |
|
294 |
-
def format_dependencies_display(dependencies_list):
|
295 |
"""Format dependencies for display."""
|
296 |
if not dependencies_list:
|
297 |
return "No dependencies added yet."
|
@@ -313,8 +314,8 @@ def format_dependencies_display(dependencies_list):
|
|
313 |
|
314 |
|
315 |
def solve_dependencies(
|
316 |
-
tasks_text, dependencies_list, durations_state:
|
317 |
-
):
|
318 |
"""Solve the task ordering problem."""
|
319 |
tasks, task_original_names = parse_tasks(tasks_text)
|
320 |
|
@@ -603,7 +604,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
603 |
""")
|
604 |
|
605 |
# Event handlers
|
606 |
-
def update_ui_after_tasks_change(tasks_text, current_deps):
|
607 |
"""Update dropdowns and dependency checkboxes when tasks change."""
|
608 |
tasks, original_names = parse_tasks(tasks_text)
|
609 |
|
@@ -626,7 +627,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
626 |
gr.CheckboxGroup(choices=checkbox_choices, value=[]), # remove_deps
|
627 |
)
|
628 |
|
629 |
-
def handle_add_dependencies(task, required_tasks, current_deps):
|
630 |
"""Handle adding dependencies and update UI."""
|
631 |
updated_deps, status = add_dependencies(task, required_tasks, current_deps)
|
632 |
display = format_dependencies_display(updated_deps)
|
@@ -640,7 +641,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
640 |
[], # requirements_checkbox (clear)
|
641 |
)
|
642 |
|
643 |
-
def handle_remove_dependencies(deps_to_remove, current_deps):
|
644 |
"""Handle removing dependencies and update UI."""
|
645 |
updated_deps, status = remove_dependency(deps_to_remove, current_deps)
|
646 |
display = format_dependencies_display(updated_deps)
|
@@ -653,7 +654,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
653 |
gr.CheckboxGroup(choices=checkbox_choices, value=[]), # remove_deps
|
654 |
)
|
655 |
|
656 |
-
def clear_all_dependencies():
|
657 |
"""Clear all dependencies."""
|
658 |
return (
|
659 |
[], # dependencies_state
|
@@ -662,7 +663,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
662 |
)
|
663 |
|
664 |
# Example loaders
|
665 |
-
def load_morning_example():
|
666 |
tasks = "Wake_up\nShower\nBrush_teeth\nGet_dressed\nEat_breakfast\nMake_coffee\nLeave_house"
|
667 |
return (
|
668 |
tasks,
|
@@ -670,7 +671,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
670 |
gr.CheckboxGroup(choices=[], value=[]),
|
671 |
)
|
672 |
|
673 |
-
def load_cooking_example():
|
674 |
tasks = "Shop_ingredients\nPrep_vegetables\nPreheat_oven\nCook_main_dish\nMake_sauce\nPlate_food\nServe_dinner"
|
675 |
return (
|
676 |
tasks,
|
@@ -678,7 +679,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
678 |
gr.CheckboxGroup(choices=[], value=[]),
|
679 |
)
|
680 |
|
681 |
-
def load_project_example():
|
682 |
tasks = "Research\nPlan\nDesign\nDevelop\nTest\nDeploy\nDocument\nReview"
|
683 |
return (
|
684 |
tasks,
|
@@ -687,7 +688,7 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
687 |
)
|
688 |
|
689 |
# Utility: update durations table to match tasks
|
690 |
-
def update_durations_table(tasks_text, old_duration_rows):
|
691 |
tasks, original_names = parse_tasks(tasks_text)
|
692 |
shown_task_set = set()
|
693 |
updated_rows = []
|
@@ -704,10 +705,10 @@ with gr.Blocks(title="Task Dependency Solver", theme=gr.themes.Soft()) as demo:
|
|
704 |
return updated_rows
|
705 |
|
706 |
# Converts Dataframe rows to durations dict (normalized)
|
707 |
-
def table_rows_to_durations(df: pandas.DataFrame):
|
708 |
return df.to_dict(orient="records")
|
709 |
|
710 |
-
def update_durations_state(rows):
|
711 |
return table_rows_to_durations(rows)
|
712 |
|
713 |
# Wire up events
|
|
|
4 |
from collections import defaultdict
|
5 |
from ortools.sat.python import cp_model
|
6 |
from datetime import datetime, timedelta
|
7 |
+
from typing import Dict, List, Tuple, Optional, Union, Any
|
8 |
|
9 |
import pandas
|
10 |
|
11 |
|
12 |
+
def normalize_task_name(task: str) -> str:
|
13 |
"""Normalize task name to lowercase for consistent comparison."""
|
14 |
return task.strip().lower().replace(" ", "_")
|
15 |
|
16 |
|
17 |
+
def display_task_name(task: str) -> str:
|
18 |
return task
|
19 |
|
20 |
|
21 |
+
def parse_requirements(requirements: List[str]) -> Tuple[Dict[str, List[str]], List[str], Dict[str, str]]:
|
22 |
"""Parse requirements list into dependency graph."""
|
23 |
dependencies = defaultdict(list)
|
24 |
all_tasks = set()
|
|
|
45 |
return dependencies, list(all_tasks), original_names
|
46 |
|
47 |
|
48 |
+
def solve_all_tasks(dependencies: Dict[str, List[str]], all_tasks: List[str]) -> Optional[List[str]]:
|
49 |
"""Try to schedule all tasks (works if DAG)."""
|
50 |
model = cp_model.CpModel()
|
51 |
n_tasks = len(all_tasks)
|
|
|
72 |
return None
|
73 |
|
74 |
|
75 |
+
def solve_maximum_subset(dependencies: Dict[str, List[str]], all_tasks: List[str]) -> List[str]:
|
76 |
"""Find maximum subset of tasks that can be executed (handles cycles)."""
|
77 |
model = cp_model.CpModel()
|
78 |
n_tasks = len(all_tasks)
|
|
|
120 |
return []
|
121 |
|
122 |
|
123 |
+
def parse_tasks(tasks_text: str) -> Tuple[List[str], Dict[str, str]]:
|
124 |
"""Parse tasks from text input."""
|
125 |
if not tasks_text.strip():
|
126 |
return [], {}
|
|
|
141 |
return sorted(tasks), original_names
|
142 |
|
143 |
|
144 |
+
def generate_mermaid_gantt(task_order: List[str], original_names: Dict[str, str], durations: Optional[Dict[str, int]] = None, start_time: Optional[datetime] = None) -> str:
|
145 |
"""Generate Mermaid Gantt chart syntax with minute granularity and selectable start time."""
|
146 |
if not task_order:
|
147 |
return "gantt\n title Task Execution Timeline\n dateFormat YYYY-MM-DD HH:mm\n section No Tasks\n No tasks to display : 2024-01-01 08:00, 1m"
|
|
|
178 |
return gantt
|
179 |
|
180 |
|
181 |
+
def generate_mermaid_flowchart(dependencies: Dict[str, List[str]], all_tasks: List[str], original_names: Dict[str, str]) -> str:
|
182 |
"""Generate Mermaid flowchart syntax."""
|
183 |
if not all_tasks:
|
184 |
return "flowchart TD\n A[No tasks to display]"
|
|
|
207 |
return flowchart
|
208 |
|
209 |
|
210 |
+
def update_dropdowns(tasks_text: str) -> Tuple[gr.Dropdown, gr.CheckboxGroup]:
|
211 |
"""Update dropdown choices when tasks change."""
|
212 |
tasks, original_names = parse_tasks(tasks_text)
|
213 |
|
|
|
225 |
)
|
226 |
|
227 |
|
228 |
+
def add_dependencies(task_display: str, required_display_list: List[str], current_deps: List[str]) -> Tuple[List[str], str]:
|
229 |
"""Add multiple dependencies for a single task."""
|
230 |
if not task_display:
|
231 |
return current_deps, "⚠️ Please select a task"
|
|
|
282 |
return updated_deps, status
|
283 |
|
284 |
|
285 |
+
def remove_dependency(deps_to_remove: List[str], current_deps: List[str]) -> Tuple[List[str], str]:
|
286 |
"""Remove selected dependencies."""
|
287 |
if not deps_to_remove:
|
288 |
return current_deps, "⚠️ Please select dependencies to remove"
|
|
|
292 |
return updated_deps, f"✅ Removed {removed_count} dependencies"
|
293 |
|
294 |
|
295 |
+
def format_dependencies_display(dependencies_list: List[str]) -> str:
|
296 |
"""Format dependencies for display."""
|
297 |
if not dependencies_list:
|
298 |
return "No dependencies added yet."
|
|
|
314 |
|
315 |
|
316 |
def solve_dependencies(
|
317 |
+
tasks_text: str, dependencies_list: List[str], durations_state: Optional[Dict[str, Union[str, int]]] = None
|
318 |
+
) -> Tuple[str, str, str, str, str, str, str]:
|
319 |
"""Solve the task ordering problem."""
|
320 |
tasks, task_original_names = parse_tasks(tasks_text)
|
321 |
|
|
|
604 |
""")
|
605 |
|
606 |
# Event handlers
|
607 |
+
def update_ui_after_tasks_change(tasks_text: str, current_deps: List[str]) -> Tuple[gr.Dropdown, gr.CheckboxGroup, gr.CheckboxGroup]:
|
608 |
"""Update dropdowns and dependency checkboxes when tasks change."""
|
609 |
tasks, original_names = parse_tasks(tasks_text)
|
610 |
|
|
|
627 |
gr.CheckboxGroup(choices=checkbox_choices, value=[]), # remove_deps
|
628 |
)
|
629 |
|
630 |
+
def handle_add_dependencies(task: str, required_tasks: List[str], current_deps: List[str]) -> Tuple[List[str], str, gr.CheckboxGroup, str, List[str]]:
|
631 |
"""Handle adding dependencies and update UI."""
|
632 |
updated_deps, status = add_dependencies(task, required_tasks, current_deps)
|
633 |
display = format_dependencies_display(updated_deps)
|
|
|
641 |
[], # requirements_checkbox (clear)
|
642 |
)
|
643 |
|
644 |
+
def handle_remove_dependencies(deps_to_remove: List[str], current_deps: List[str]) -> Tuple[List[str], str, str, gr.CheckboxGroup]:
|
645 |
"""Handle removing dependencies and update UI."""
|
646 |
updated_deps, status = remove_dependency(deps_to_remove, current_deps)
|
647 |
display = format_dependencies_display(updated_deps)
|
|
|
654 |
gr.CheckboxGroup(choices=checkbox_choices, value=[]), # remove_deps
|
655 |
)
|
656 |
|
657 |
+
def clear_all_dependencies() -> Tuple[List[str], str, gr.CheckboxGroup]:
|
658 |
"""Clear all dependencies."""
|
659 |
return (
|
660 |
[], # dependencies_state
|
|
|
663 |
)
|
664 |
|
665 |
# Example loaders
|
666 |
+
def load_morning_example() -> Tuple[str, List[str], gr.CheckboxGroup]:
|
667 |
tasks = "Wake_up\nShower\nBrush_teeth\nGet_dressed\nEat_breakfast\nMake_coffee\nLeave_house"
|
668 |
return (
|
669 |
tasks,
|
|
|
671 |
gr.CheckboxGroup(choices=[], value=[]),
|
672 |
)
|
673 |
|
674 |
+
def load_cooking_example() -> Tuple[str, List[str], gr.CheckboxGroup]:
|
675 |
tasks = "Shop_ingredients\nPrep_vegetables\nPreheat_oven\nCook_main_dish\nMake_sauce\nPlate_food\nServe_dinner"
|
676 |
return (
|
677 |
tasks,
|
|
|
679 |
gr.CheckboxGroup(choices=[], value=[]),
|
680 |
)
|
681 |
|
682 |
+
def load_project_example() -> Tuple[str, List[str], gr.CheckboxGroup]:
|
683 |
tasks = "Research\nPlan\nDesign\nDevelop\nTest\nDeploy\nDocument\nReview"
|
684 |
return (
|
685 |
tasks,
|
|
|
688 |
)
|
689 |
|
690 |
# Utility: update durations table to match tasks
|
691 |
+
def update_durations_table(tasks_text: str, old_duration_rows: List[List[Any]]) -> List[List[str]]:
|
692 |
tasks, original_names = parse_tasks(tasks_text)
|
693 |
shown_task_set = set()
|
694 |
updated_rows = []
|
|
|
705 |
return updated_rows
|
706 |
|
707 |
# Converts Dataframe rows to durations dict (normalized)
|
708 |
+
def table_rows_to_durations(df: pandas.DataFrame) -> List[Dict[str, Any]]:
|
709 |
return df.to_dict(orient="records")
|
710 |
|
711 |
+
def update_durations_state(rows: pandas.DataFrame) -> List[Dict[str, Any]]:
|
712 |
return table_rows_to_durations(rows)
|
713 |
|
714 |
# Wire up events
|
schedule.py
CHANGED
@@ -1,6 +1,6 @@
|
|
1 |
from ortools.sat.python import cp_model
|
2 |
|
3 |
-
def main():
|
4 |
model = cp_model.CpModel()
|
5 |
horizon = 48 # 24-hour day in 30-minute slots (adjust as needed)
|
6 |
|
|
|
1 |
from ortools.sat.python import cp_model
|
2 |
|
3 |
+
def main() -> None:
|
4 |
model = cp_model.CpModel()
|
5 |
horizon = 48 # 24-hour day in 30-minute slots (adjust as needed)
|
6 |
|
test_app.py
CHANGED
@@ -2,7 +2,7 @@ from app import parse_requirements, parse_tasks, solve_all_tasks, generate_merma
|
|
2 |
from datetime import datetime
|
3 |
|
4 |
|
5 |
-
def test_parse_requirements_with_spaces():
|
6 |
reqs = [
|
7 |
"take out trash requires bag",
|
8 |
"Dinner requires Shopping",
|
@@ -23,7 +23,7 @@ def test_parse_requirements_with_spaces():
|
|
23 |
assert original_names["take out trash"] == "take out trash"
|
24 |
|
25 |
|
26 |
-
def test_parse_tasks_and_original_names():
|
27 |
txt = "Wash dishes\nTake out trash, Make sandwich "
|
28 |
tasks, originals = parse_tasks(txt)
|
29 |
assert set(tasks) == {"wash dishes", "take out trash", "make sandwich"}
|
@@ -32,14 +32,14 @@ def test_parse_tasks_and_original_names():
|
|
32 |
assert originals["wash dishes"] == "Wash dishes"
|
33 |
|
34 |
|
35 |
-
def test_solve_all_tasks_simple():
|
36 |
reqs = ["a requires b", "b requires c"]
|
37 |
dependencies, all_tasks, _ = parse_requirements(reqs)
|
38 |
result = solve_all_tasks(dependencies, all_tasks)
|
39 |
assert result == ["c", "b", "a"]
|
40 |
|
41 |
|
42 |
-
def test_solve_all_tasks_with_spaces():
|
43 |
reqs = ["Do homework requires eat dinner", "eat dinner requires cook"]
|
44 |
dependencies, all_tasks, _ = parse_requirements(reqs)
|
45 |
result = solve_all_tasks(dependencies, all_tasks)
|
@@ -54,7 +54,7 @@ def test_solve_all_tasks_with_spaces():
|
|
54 |
# assert result == ["d"]
|
55 |
|
56 |
|
57 |
-
def test_generate_mermaid_gantt_minutes():
|
58 |
order = ["a", "b", "c"]
|
59 |
names = {"a": "Alpha", "b": "Beta", "c": "Gamma"}
|
60 |
durations = {"a": 15, "b": 30, "c": 45}
|
|
|
2 |
from datetime import datetime
|
3 |
|
4 |
|
5 |
+
def test_parse_requirements_with_spaces() -> None:
|
6 |
reqs = [
|
7 |
"take out trash requires bag",
|
8 |
"Dinner requires Shopping",
|
|
|
23 |
assert original_names["take out trash"] == "take out trash"
|
24 |
|
25 |
|
26 |
+
def test_parse_tasks_and_original_names() -> None:
|
27 |
txt = "Wash dishes\nTake out trash, Make sandwich "
|
28 |
tasks, originals = parse_tasks(txt)
|
29 |
assert set(tasks) == {"wash dishes", "take out trash", "make sandwich"}
|
|
|
32 |
assert originals["wash dishes"] == "Wash dishes"
|
33 |
|
34 |
|
35 |
+
def test_solve_all_tasks_simple() -> None:
|
36 |
reqs = ["a requires b", "b requires c"]
|
37 |
dependencies, all_tasks, _ = parse_requirements(reqs)
|
38 |
result = solve_all_tasks(dependencies, all_tasks)
|
39 |
assert result == ["c", "b", "a"]
|
40 |
|
41 |
|
42 |
+
def test_solve_all_tasks_with_spaces() -> None:
|
43 |
reqs = ["Do homework requires eat dinner", "eat dinner requires cook"]
|
44 |
dependencies, all_tasks, _ = parse_requirements(reqs)
|
45 |
result = solve_all_tasks(dependencies, all_tasks)
|
|
|
54 |
# assert result == ["d"]
|
55 |
|
56 |
|
57 |
+
def test_generate_mermaid_gantt_minutes() -> None:
|
58 |
order = ["a", "b", "c"]
|
59 |
names = {"a": "Alpha", "b": "Beta", "c": "Gamma"}
|
60 |
durations = {"a": 15, "b": 30, "c": 45}
|