Spaces:
Sleeping
Sleeping
File size: 3,646 Bytes
ee2322f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
import random
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any
PRIORITIES = ["P0", "P1", "P2", "P3"]
@dataclass
class Task:
task_id: str
title: str
task_type: str
priority: str
created_t: int
sla_due_t: int
est_effort_min: int
value_usd: float
dependencies: List[str] = field(default_factory=list)
status: str = "QUEUED" # QUEUED | IN_PROGRESS | BLOCKED | DONE | REWORK
owner: Optional[str] = None
started_t: Optional[int] = None
completed_t: Optional[int] = None
attempts: int = 0
notes: Dict[str, Any] = field(default_factory=dict)
class TaskSystem:
def __init__(self, seed: int):
self.rng = random.Random(seed)
self.tasks: Dict[str, Task] = {}
self.counter = 0
def _next_id(self) -> str:
self.counter += 1
return f"T{self.counter:05d}"
def create_task(
self,
t_sim: int,
title: str,
task_type: str,
priority: str,
sla_ticks: int,
est_effort_min: int,
value_usd: float,
dependencies: Optional[List[str]] = None,
urgent: bool = False,
) -> Task:
tid = self._next_id()
sla_due = t_sim + max(1, int(sla_ticks))
if urgent:
priority = "P0"
task = Task(
task_id=tid,
title=title,
task_type=task_type,
priority=priority,
created_t=t_sim,
sla_due_t=sla_due,
est_effort_min=est_effort_min,
value_usd=value_usd,
dependencies=dependencies or [],
)
self.tasks[tid] = task
return task
def queued_tasks(self) -> List[Task]:
return [t for t in self.tasks.values() if t.status in ("QUEUED", "REWORK")]
def active_tasks(self) -> List[Task]:
return [t for t in self.tasks.values() if t.status in ("IN_PROGRESS", "BLOCKED")]
def done_tasks(self) -> List[Task]:
return [t for t in self.tasks.values() if t.status == "DONE"]
def overdue_tasks(self, t_sim: int) -> List[Task]:
return [t for t in self.tasks.values() if t.status != "DONE" and t_sim > t.sla_due_t]
def can_start(self, task: Task) -> bool:
# Dependencies must be DONE
for dep in task.dependencies:
if dep in self.tasks and self.tasks[dep].status != "DONE":
return False
return True
def pick_next_task(self, t_sim: int) -> Optional[Task]:
# Priority then due date
q = [t for t in self.queued_tasks() if self.can_start(t)]
if not q:
return None
pr_order = {p: i for i, p in enumerate(PRIORITIES)}
q.sort(key=lambda t: (pr_order.get(t.priority, 9), t.sla_due_t, t.created_t))
return q[0]
def to_compact_table(self, t_sim: int, limit: int = 14) -> str:
# Simple readable view for UI
items = sorted(self.tasks.values(), key=lambda t: (t.status != "DONE", t.sla_due_t))
items = items[:limit]
lines = ["id | pri | status | due(t) | owner | title"]
lines.append("---|-----|--------|--------|-------|------")
for t in items:
due = t.sla_due_t
lines.append(f"{t.task_id} | {t.priority} | {t.status} | {due} | {t.owner or '-'} | {t.title[:42]}")
overdue = len(self.overdue_tasks(t_sim))
lines.append(f"\nOverdue: {overdue} | Total: {len(self.tasks)} | Done: {len(self.done_tasks())}")
return "\n".join(lines)
def as_dict(self) -> Dict[str, Any]:
return {k: asdict(v) for k, v in self.tasks.items()}
|