90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import PomodoroCycle
|
|
|
|
PHASE_WORK = "work"
|
|
PHASE_SHORT_BREAK = "short_break"
|
|
PHASE_LONG_BREAK = "long_break"
|
|
|
|
|
|
class CycleManager:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get(self) -> PomodoroCycle:
|
|
cycle = self.db.scalar(select(PomodoroCycle).limit(1))
|
|
if not cycle:
|
|
cycle = PomodoroCycle()
|
|
self.db.add(cycle)
|
|
self.db.commit()
|
|
self.db.refresh(cycle)
|
|
return cycle
|
|
|
|
def to_dict(self, cycle: PomodoroCycle | None = None) -> dict:
|
|
c = cycle or self.get()
|
|
return {
|
|
"completed_work_sessions": c.completed_work_sessions,
|
|
"sessions_until_long_break": c.sessions_until_long_break,
|
|
"task_note": c.task_note,
|
|
"work_duration_min": c.work_duration_min,
|
|
"short_break_min": c.short_break_min,
|
|
"long_break_min": c.long_break_min,
|
|
"auto_advance": c.auto_advance,
|
|
"chat_notify_seq": c.chat_notify_seq,
|
|
}
|
|
|
|
def reset(self, clear_task: bool = False) -> dict:
|
|
cycle = self.get()
|
|
cycle.completed_work_sessions = 0
|
|
if clear_task:
|
|
cycle.task_note = ""
|
|
self.db.commit()
|
|
self.db.refresh(cycle)
|
|
return self.to_dict(cycle)
|
|
|
|
def bump_notify_seq(self) -> int:
|
|
cycle = self.get()
|
|
cycle.chat_notify_seq += 1
|
|
self.db.commit()
|
|
self.db.refresh(cycle)
|
|
return cycle.chat_notify_seq
|
|
|
|
def on_work_completed(self) -> str:
|
|
"""Returns next phase: short_break or long_break."""
|
|
cycle = self.get()
|
|
cycle.completed_work_sessions += 1
|
|
if cycle.completed_work_sessions >= cycle.sessions_until_long_break:
|
|
next_phase = PHASE_LONG_BREAK
|
|
else:
|
|
next_phase = PHASE_SHORT_BREAK
|
|
self.db.commit()
|
|
return next_phase
|
|
|
|
def on_long_break_completed(self) -> None:
|
|
cycle = self.get()
|
|
cycle.completed_work_sessions = 0
|
|
self.db.commit()
|
|
|
|
def duration_for_phase(self, phase: str, cycle: PomodoroCycle | None = None) -> int:
|
|
c = cycle or self.get()
|
|
if phase == PHASE_WORK:
|
|
return c.work_duration_min
|
|
if phase == PHASE_SHORT_BREAK:
|
|
return c.short_break_min
|
|
if phase == PHASE_LONG_BREAK:
|
|
return c.long_break_min
|
|
return c.work_duration_min
|
|
|
|
def next_phase_after(self, completed_phase: str) -> str | None:
|
|
if completed_phase == PHASE_WORK:
|
|
cycle = self.get()
|
|
if cycle.completed_work_sessions >= cycle.sessions_until_long_break:
|
|
return PHASE_LONG_BREAK
|
|
return PHASE_SHORT_BREAK
|
|
if completed_phase == PHASE_SHORT_BREAK:
|
|
return PHASE_WORK
|
|
if completed_phase == PHASE_LONG_BREAK:
|
|
return None
|
|
return None
|