timka123
77b26757bf
- Refactored TimerPage component to improve layout and user experience. - Updated theme configuration to utilize design tokens for better maintainability. - Introduced new tokens for colors and typography to align with design principles. - Simplified state management and improved timer functionality. - Enhanced accessibility and responsiveness of UI elements.
283 lines
11 KiB
Python
283 lines
11 KiB
Python
"""Turning rows into numbers: a day, a range, streaks, and a heatmap.
|
|
|
|
Everything the report and the dashboard show comes from here, so the web UI and
|
|
the Telegram message can never disagree about how many pomodoros you did.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import BlockEvent, PomodoroSession, TaskRecord
|
|
from app.settings_schema import AppSettings
|
|
from app.timeutil import local_date, to_local
|
|
|
|
# Two work intervals closer together than this count as one unbroken stretch.
|
|
_STREAK_GAP_SECONDS = 20 * 60
|
|
|
|
|
|
def _focus_blocks(rows: list[PomodoroSession], tz_name: str) -> list[tuple[datetime, datetime]]:
|
|
blocks: list[tuple[datetime, datetime]] = []
|
|
for row in rows:
|
|
if row.kind != "work" or row.status != "completed" or row.ended_at is None:
|
|
continue
|
|
blocks.append((to_local(row.started_at, tz_name), to_local(row.ended_at, tz_name)))
|
|
return sorted(blocks)
|
|
|
|
|
|
def longest_focus_stretch_minutes(rows: list[PomodoroSession], tz_name: str) -> int:
|
|
"""The longest run of back-to-back pomodoros, in minutes of actual focus."""
|
|
blocks = _focus_blocks(rows, tz_name)
|
|
if not blocks:
|
|
return 0
|
|
best = current = (blocks[0][1] - blocks[0][0]).total_seconds()
|
|
previous_end = blocks[0][1]
|
|
for start, end in blocks[1:]:
|
|
gap = (start - previous_end).total_seconds()
|
|
length = (end - start).total_seconds()
|
|
current = current + length if gap <= _STREAK_GAP_SECONDS else length
|
|
best = max(best, current)
|
|
previous_end = end
|
|
return int(best // 60)
|
|
|
|
|
|
def day_metrics(session: Session, settings: AppSettings, day: date) -> dict[str, Any]:
|
|
"""Everything measurable about one day."""
|
|
tz_name = settings.timezone
|
|
rows = list(
|
|
session.scalars(
|
|
select(PomodoroSession)
|
|
.where(PomodoroSession.local_date == day)
|
|
.order_by(PomodoroSession.started_at.asc())
|
|
)
|
|
)
|
|
|
|
work_done = [r for r in rows if r.kind == "work" and r.status == "completed"]
|
|
work_dropped = [r for r in rows if r.kind == "work" and r.status == "abandoned"]
|
|
breaks_done = [r for r in rows if r.kind != "work" and r.status == "completed"]
|
|
|
|
focus_seconds = sum(r.actual_seconds for r in work_done)
|
|
break_seconds = sum(r.actual_seconds for r in breaks_done)
|
|
dropped_seconds = sum(r.actual_seconds for r in work_dropped)
|
|
|
|
by_hour = Counter(to_local(r.started_at, tz_name).hour for r in work_done)
|
|
# Closed in Jira / Todoist / Google and noticed by a refresh — this app
|
|
# never ticks a task off itself.
|
|
tasks = list(
|
|
session.scalars(select(TaskRecord).where(TaskRecord.completed_date == day))
|
|
)
|
|
|
|
blocks = list(
|
|
session.scalars(select(BlockEvent).where(BlockEvent.local_date == day))
|
|
)
|
|
block_counter = Counter(event.target for event in blocks)
|
|
|
|
pomodoro_time_by_task: Counter[str] = Counter()
|
|
for row in work_done:
|
|
label = row.task_title or "Unassigned"
|
|
pomodoro_time_by_task[label] += row.actual_seconds
|
|
|
|
first = to_local(work_done[0].started_at, tz_name) if work_done else None
|
|
last_row = work_done[-1] if work_done else None
|
|
last = to_local(last_row.ended_at or last_row.started_at, tz_name) if last_row else None
|
|
|
|
goal = settings.pomodoro.daily_goal_pomodoros
|
|
completed = len(work_done)
|
|
planned_seconds = sum(r.planned_seconds for r in work_done)
|
|
|
|
return {
|
|
"date": day.isoformat(),
|
|
"pomodoros": completed,
|
|
"goal": goal,
|
|
"goal_pct": round(100 * completed / goal) if goal else 0,
|
|
"goal_met": completed >= goal,
|
|
"focus_minutes": round(focus_seconds / 60),
|
|
"break_minutes": round(break_seconds / 60),
|
|
"wasted_minutes": round(dropped_seconds / 60),
|
|
"abandoned": len(work_dropped),
|
|
"abandon_reasons": [r.abandon_reason for r in work_dropped if r.abandon_reason],
|
|
"breaks_taken": len(breaks_done),
|
|
"breaks_owed": completed,
|
|
"interruptions": sum(r.interruptions for r in rows),
|
|
"completion_rate": (
|
|
round(100 * focus_seconds / planned_seconds) if planned_seconds else 0
|
|
),
|
|
"longest_stretch_minutes": longest_focus_stretch_minutes(rows, tz_name),
|
|
"first_start": first.strftime("%H:%M") if first else None,
|
|
"last_end": last.strftime("%H:%M") if last else None,
|
|
"by_hour": [{"hour": hour, "count": by_hour.get(hour, 0)} for hour in range(24)],
|
|
"peak_hour": (by_hour.most_common(1)[0][0] if by_hour else None),
|
|
"tasks_completed": [
|
|
{
|
|
"id": task.id,
|
|
"title": task.title,
|
|
"source": task.source,
|
|
"project": task.project,
|
|
"url": task.url,
|
|
}
|
|
for task in tasks
|
|
],
|
|
"tasks_completed_count": len(tasks),
|
|
"time_by_task": [
|
|
{"title": title, "minutes": round(seconds / 60)}
|
|
for title, seconds in pomodoro_time_by_task.most_common(10)
|
|
],
|
|
"blocked_attempts": sum(block_counter.values()),
|
|
"blocked_top": [
|
|
{"target": target, "count": count} for target, count in block_counter.most_common(8)
|
|
],
|
|
"sessions": [
|
|
{
|
|
"id": row.id,
|
|
"kind": row.kind,
|
|
"status": row.status,
|
|
"started_at": to_local(row.started_at, tz_name).isoformat(),
|
|
"minutes": round(row.actual_seconds / 60),
|
|
"task": row.task_title,
|
|
"interruptions": row.interruptions,
|
|
}
|
|
for row in rows
|
|
],
|
|
}
|
|
|
|
|
|
def range_metrics(
|
|
session: Session, settings: AppSettings, start: date, end: date
|
|
) -> list[dict[str, Any]]:
|
|
"""Per-day metrics for `[start, end]`, inclusive, cheaply (one query)."""
|
|
rows = list(
|
|
session.scalars(
|
|
select(PomodoroSession).where(
|
|
PomodoroSession.local_date >= start,
|
|
PomodoroSession.local_date <= end,
|
|
PomodoroSession.kind == "work",
|
|
PomodoroSession.status == "completed",
|
|
)
|
|
)
|
|
)
|
|
per_day: dict[date, list[PomodoroSession]] = {}
|
|
for row in rows:
|
|
per_day.setdefault(row.local_date, []).append(row)
|
|
|
|
out: list[dict[str, Any]] = []
|
|
cursor = start
|
|
while cursor <= end:
|
|
day_rows = per_day.get(cursor, [])
|
|
out.append(
|
|
{
|
|
"date": cursor.isoformat(),
|
|
"pomodoros": len(day_rows),
|
|
"focus_minutes": round(sum(r.actual_seconds for r in day_rows) / 60),
|
|
"goal_met": len(day_rows) >= settings.pomodoro.daily_goal_pomodoros,
|
|
}
|
|
)
|
|
cursor += timedelta(days=1)
|
|
return out
|
|
|
|
|
|
def averages(session: Session, settings: AppSettings, days: int, *, until: date) -> dict[str, float]:
|
|
"""Mean pomodoros/focus over the `days` days ending the day before `until`."""
|
|
start = until - timedelta(days=days)
|
|
end = until - timedelta(days=1)
|
|
if end < start:
|
|
return {"pomodoros": 0.0, "focus_minutes": 0.0}
|
|
rows = range_metrics(session, settings, start, end)
|
|
if not rows:
|
|
return {"pomodoros": 0.0, "focus_minutes": 0.0}
|
|
return {
|
|
"pomodoros": round(sum(r["pomodoros"] for r in rows) / len(rows), 1),
|
|
"focus_minutes": round(sum(r["focus_minutes"] for r in rows) / len(rows), 1),
|
|
}
|
|
|
|
|
|
def streak(session: Session, settings: AppSettings, *, until: date | None = None) -> dict[str, int]:
|
|
"""Consecutive days ending at `until` with at least one pomodoro, and with the goal met.
|
|
|
|
Today only breaks a streak once it is over, so a day with nothing on it yet
|
|
keeps yesterday's streak alive.
|
|
"""
|
|
until = until or local_date(settings.timezone)
|
|
days = list(
|
|
session.scalars(
|
|
select(PomodoroSession.local_date)
|
|
.where(
|
|
PomodoroSession.kind == "work",
|
|
PomodoroSession.status == "completed",
|
|
PomodoroSession.local_date <= until,
|
|
)
|
|
.group_by(PomodoroSession.local_date)
|
|
)
|
|
)
|
|
counts = dict(
|
|
session.execute(
|
|
select(PomodoroSession.local_date, func.count(PomodoroSession.id))
|
|
.where(
|
|
PomodoroSession.kind == "work",
|
|
PomodoroSession.status == "completed",
|
|
PomodoroSession.local_date <= until,
|
|
)
|
|
.group_by(PomodoroSession.local_date)
|
|
).all()
|
|
)
|
|
active = set(days)
|
|
|
|
def run(from_day: date, predicate) -> int:
|
|
length, cursor = 0, from_day
|
|
while predicate(cursor):
|
|
length += 1
|
|
cursor -= timedelta(days=1)
|
|
return length
|
|
|
|
goal = settings.pomodoro.daily_goal_pomodoros
|
|
anchor = until if until in active else until - timedelta(days=1)
|
|
goal_anchor = (
|
|
until if counts.get(until, 0) >= goal else until - timedelta(days=1)
|
|
)
|
|
return {
|
|
"current": run(anchor, lambda day: day in active),
|
|
"goal_streak": run(goal_anchor, lambda day: counts.get(day, 0) >= goal),
|
|
"best": _best_streak(sorted(active)),
|
|
"total_days": len(active),
|
|
}
|
|
|
|
|
|
def _best_streak(days: list[date]) -> int:
|
|
best = run = 0
|
|
previous: date | None = None
|
|
for day in days:
|
|
run = run + 1 if previous is not None and (day - previous).days == 1 else 1
|
|
best = max(best, run)
|
|
previous = day
|
|
return best
|
|
|
|
|
|
def heatmap(session: Session, settings: AppSettings, weeks: int = 12) -> list[dict[str, Any]]:
|
|
"""Per-day counts for the last `weeks` weeks, for the dashboard's calendar."""
|
|
end = local_date(settings.timezone)
|
|
start = end - timedelta(weeks=weeks)
|
|
return range_metrics(session, settings, start, end)
|
|
|
|
|
|
def lifetime(session: Session, settings: AppSettings) -> dict[str, Any]:
|
|
total, seconds = session.execute(
|
|
select(func.count(PomodoroSession.id), func.coalesce(func.sum(PomodoroSession.actual_seconds), 0)).where(
|
|
PomodoroSession.kind == "work", PomodoroSession.status == "completed"
|
|
)
|
|
).one()
|
|
tasks = session.scalar(
|
|
select(func.count(TaskRecord.id)).where(TaskRecord.completed_date.is_not(None))
|
|
)
|
|
blocked = session.scalar(select(func.count(BlockEvent.id)))
|
|
return {
|
|
"pomodoros": int(total or 0),
|
|
"focus_minutes": round(int(seconds or 0) / 60),
|
|
"focus_hours": round(int(seconds or 0) / 3600, 1),
|
|
"tasks_completed": int(tasks or 0),
|
|
"blocked_attempts": int(blocked or 0),
|
|
}
|