96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
"""Kanban board logic for the dashboard."""
|
|
|
|
from collections import defaultdict
|
|
from typing import Optional
|
|
from .task import Task, TaskState, COLUMN_HEADERS
|
|
|
|
|
|
class KanbanBoard:
|
|
# Grouping configuration
|
|
GROUPS = {
|
|
"Planning": [
|
|
TaskState.BACKLOG, TaskState.RESEARCH, TaskState.DECOMPOSITION
|
|
],
|
|
"Design": [
|
|
TaskState.DESIGN, TaskState.TEST_DESIGN
|
|
],
|
|
"Implementation": [
|
|
TaskState.IMPLEMENT
|
|
],
|
|
"Verification": [
|
|
TaskState.BUG_FIND, TaskState.ADV_BUG_FIND, TaskState.DOC_REVIEW, TaskState.REFEREE
|
|
],
|
|
"Blocked": [
|
|
TaskState.BLOCKED
|
|
],
|
|
"Resolution": [
|
|
TaskState.DONE
|
|
],
|
|
}
|
|
|
|
def _build_columns(self) -> dict[TaskState, list[Task]]:
|
|
columns = {state: [] for state in self.COLUMNS}
|
|
for task in self.tasks:
|
|
if task.state in columns:
|
|
columns[task.state].append(task)
|
|
else:
|
|
columns[TaskState.BACKLOG].append(task)
|
|
return columns
|
|
|
|
@property
|
|
def columns(self) -> dict[TaskState, list[Task]]:
|
|
return self._columns
|
|
|
|
@property
|
|
def column_order(self) -> list[TaskState]:
|
|
return self.COLUMNS
|
|
|
|
def get_grouped_columns(self) -> dict[str, list[Task]]:
|
|
grouped = {group_name: [] for group_name in self.GROUPS.keys()}
|
|
for state, tasks in self._columns.items():
|
|
for group_name, states in self.GROUPS.items():
|
|
if state in states:
|
|
grouped[group_name].extend(tasks)
|
|
break
|
|
return grouped
|
|
|
|
@property
|
|
def total_tasks(self) -> int:
|
|
return len(self.tasks)
|
|
|
|
def get_column_width(self, col: TaskState) -> int:
|
|
header = COLUMN_HEADERS.get(col, col.value)
|
|
return max(self.column_width, len(header) + 4)
|
|
|
|
def filter_columns(
|
|
self, phase_filter: Optional[TaskState] = None,
|
|
wave_filter: Optional[str] = None, search_query: Optional[str] = None,
|
|
) -> dict[TaskState, list[Task]]:
|
|
filtered = {state: [] for state in self.COLUMNS}
|
|
for task in self.tasks:
|
|
if phase_filter and task.state != phase_filter:
|
|
continue
|
|
if wave_filter == "no-waves" and task.sub_tasks:
|
|
continue
|
|
elif wave_filter == "has-waves" and not task.sub_tasks:
|
|
continue
|
|
if search_query:
|
|
query_lower = search_query.lower()
|
|
name_match = query_lower in task.name.lower() or query_lower in task.display_name.lower()
|
|
subtask_match = any(query_lower in st.name.lower() for st in task.sub_tasks)
|
|
if not name_match and not subtask_match:
|
|
continue
|
|
filtered[task.state].append(task)
|
|
return filtered
|
|
|
|
def get_tasks_by_state(self, state: TaskState) -> list[Task]:
|
|
return self._columns.get(state, [])
|
|
|
|
def get_wip_count(self) -> int:
|
|
wip_states = [
|
|
TaskState.RESEARCH, TaskState.DECOMPOSITION, TaskState.DESIGN,
|
|
TaskState.TEST_DESIGN, TaskState.IMPLEMENT, TaskState.BUG_FIND,
|
|
TaskState.ADV_BUG_FIND, TaskState.DOC_REVIEW, TaskState.REFEREE,
|
|
]
|
|
return sum(len(self._columns.get(s, [])) for s in wip_states)
|