"""Actionables."""
from __future__ import annotations
-from dataclasses import dataclass
-from typing import Any
+from typing import Any, Set
from sqlite3 import Row
+from plomtask.misc import DictableNode
from plomtask.db import DatabaseConnection, BaseModel
from plomtask.processes import Process, ProcessStepsNode
from plomtask.versioned_attributes import VersionedAttribute
from plomtask.dating import valid_date
-@dataclass
-class TodoNode:
+class TodoNode(DictableNode):
"""Collects what's useful to know for Todo/Condition tree display."""
+ # pylint: disable=too-few-public-methods
todo: Todo
seen: bool
children: list[TodoNode]
+ _to_dict = ['todo', 'seen', 'children']
+
+
+class TodoOrProcStepNode(DictableNode):
+ """Collect what's useful for Todo-or-ProcessStep tree display."""
+ # pylint: disable=too-few-public-methods
+ node_id: int
+ todo: Todo | None
+ process: Process | None
+ children: list[TodoOrProcStepNode] # pylint: disable=undefined-variable
+ fillable: bool = False
+ _to_dict = ['node_id', 'todo', 'process', 'children', 'fillable']
class Todo(BaseModel[int], ConditionsRelations):
"""Individual actionable."""
# pylint: disable=too-many-instance-attributes
+ # pylint: disable=too-many-public-methods
table_name = 'todos'
- to_save = ['process_id', 'is_done', 'date', 'comment', 'effort',
- 'calendarize']
+ to_save_simples = ['process_id', 'is_done', 'date', 'comment', 'effort',
+ 'calendarize']
to_save_relations = [('todo_conditions', 'todo', 'conditions', 0),
('todo_blockers', 'todo', 'blockers', 0),
('todo_enables', 'todo', 'enables', 0),
('todo_children', 'parent', 'children', 0),
('todo_children', 'child', 'parents', 1)]
to_search = ['comment']
+ days_to_update: Set[str] = set()
+ children: list[Todo]
+ parents: list[Todo]
+ sorters = {'doneness': lambda t: t.is_done,
+ 'title': lambda t: t.title_then,
+ 'comment': lambda t: t.comment,
+ 'date': lambda t: t.date}
# pylint: disable=too-many-arguments
def __init__(self, id_: int | None,
self.date = valid_date(date)
self.comment = comment
self.effort = effort
- self.children: list[Todo] = []
- self.parents: list[Todo] = []
+ self.children = []
+ self.parents = []
self.calendarize = calendarize
if not self.id_:
self.calendarize = self.process.calendarize
todos, _, _ = cls.by_date_range_with_limits(db_conn, date_range)
return todos
- @classmethod
- def create_with_children(cls, db_conn: DatabaseConnection,
- process_id: int, date: str) -> Todo:
- """Create Todo of process for date, ensure children."""
-
- def key_order_func(n: ProcessStepsNode) -> int:
- assert isinstance(n.process.id_, int)
- return n.process.id_
+ def ensure_children(self, db_conn: DatabaseConnection) -> None:
+ """Ensure Todo children (create or adopt) demanded by Process chain."""
def walk_steps(parent: Todo, step_node: ProcessStepsNode) -> Todo:
- adoptables = [t for t in cls.by_date(db_conn, date)
+ adoptables = [t for t in Todo.by_date(db_conn, parent.date)
if (t not in parent.children)
and (t != parent)
- and step_node.process == t.process]
+ and step_node.process.id_ == t.process_id]
satisfier = None
for adoptable in adoptables:
satisfier = adoptable
break
if not satisfier:
- satisfier = cls(None, step_node.process, False, date)
+ satisfier = Todo(None, step_node.process, False, parent.date)
satisfier.save(db_conn)
- sub_step_nodes = list(step_node.steps.values())
- sub_step_nodes.sort(key=key_order_func)
+ sub_step_nodes = sorted(
+ step_node.steps,
+ key=lambda s: s.process.id_ if s.process.id_ else 0)
for sub_node in sub_step_nodes:
+ if sub_node.is_suppressed:
+ continue
n_slots = len([n for n in sub_step_nodes
if n.process == sub_node.process])
filled_slots = len([t for t in satisfier.children
- if t.process == sub_node.process])
+ if t.process.id_ == sub_node.process.id_])
# if we did not newly create satisfier, it may already fill
# some step dependencies, so only fill what remains open
if n_slots - filled_slots > 0:
satisfier.save(db_conn)
return satisfier
- process = Process.by_id(db_conn, process_id)
- todo = cls(None, process, False, date)
- todo.save(db_conn)
+ process = Process.by_id(db_conn, self.process_id)
steps_tree = process.get_steps(db_conn)
- for step_node in steps_tree.values():
- todo.add_child(walk_steps(todo, step_node))
- todo.save(db_conn)
- return todo
+ for step_node in steps_tree:
+ if step_node.is_suppressed:
+ continue
+ self.add_child(walk_steps(self, step_node))
+ self.save(db_conn)
@classmethod
def from_table_row(cls, db_conn: DatabaseConnection,
assert isinstance(todo.id_, int)
for t_id in db_conn.column_where('todo_children', 'child',
'parent', todo.id_):
- # pylint: disable=no-member
todo.children += [cls.by_id(db_conn, t_id)]
for t_id in db_conn.column_where('todo_children', 'parent',
'child', todo.id_):
- # pylint: disable=no-member
todo.parents += [cls.by_id(db_conn, t_id)]
for name in ('conditions', 'blockers', 'enables', 'disables'):
table = f'todo_{name}'
return True
@property
- def process_id(self) -> int | str | None:
+ def performed_effort(self) -> float:
+ """Return performed effort, i.e. self.effort or default if done.."""
+ if self.effort is not None:
+ return self.effort
+ if self.is_done:
+ return self.effort_then
+ return 0
+
+ @property
+ def process_id(self) -> int:
"""Needed for super().save to save Processes as attributes."""
+ assert isinstance(self.process.id_, int)
return self.process.id_
@property
@property
def title(self) -> VersionedAttribute:
"""Shortcut to .process.title."""
+ assert isinstance(self.process.title, VersionedAttribute)
return self.process.title
@property
assert isinstance(effort_then, float)
return effort_then
+ @property
+ def has_doneness_in_path(self) -> bool:
+ """Check whether self is done or has any children that are."""
+ if self.is_done:
+ return True
+ for child in self.children:
+ if child.is_done:
+ return True
+ if child.has_doneness_in_path:
+ return True
+ return False
+
def get_step_tree(self, seen_todos: set[int]) -> TodoNode:
"""Return tree of depended-on Todos."""
return make_node(self)
+ @property
+ def tree_effort(self) -> float:
+ """Return sum of performed efforts of self and all descendants."""
+
+ def walk_tree(node: Todo) -> float:
+ local_effort = 0.0
+ for child in node.children:
+ local_effort += walk_tree(child)
+ return node.performed_effort + local_effort
+
+ return walk_tree(self)
+
def add_child(self, child: Todo) -> None:
"""Add child to self.children, avoid recursion, update parenthoods."""
self.children.remove(child)
child.parents.remove(self)
+ def update_attrs(self, **kwargs: Any) -> None:
+ """Update self's attributes listed in kwargs."""
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+
def save(self, db_conn: DatabaseConnection) -> None:
"""On save calls, also check if auto-deletion by effort < 0."""
if self.effort and self.effort < 0 and self.is_deletable:
self.remove(db_conn)
return
+ if self.id_ is None:
+ self.__class__.days_to_update.add(self.date)
super().save(db_conn)
+ for condition in self.enables + self.disables + self.conditions:
+ condition.save(db_conn)
def remove(self, db_conn: DatabaseConnection) -> None:
"""Remove from DB, including relations."""
if not self.is_deletable:
raise HandledException('Cannot remove non-deletable Todo.')
+ self.__class__.days_to_update.add(self.date)
children_to_remove = self.children[:]
parents_to_remove = self.parents[:]
for child in children_to_remove:
projects / plomtask / blobdiff