screen-locker/screen_locker/_status_data.py

"""Read-only status snapshot: today, this week, shutdown projection, sick budget.

Every function in this module only reads files already on disk — no ADB, no
sudo, no network, no writes. ``_status.py`` stays untouched (it's an
existing, tested, deliberately *mutating* CLI used by a human running
``--status`` who already expects that); this module is the new, genuinely
side-effect-free layer that both the i3blocks summary line and the Tkinter
status view are built on.
"""

from __future__ import annotations

from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import TYPE_CHECKING

from screen_locker import _manual_workout
from screen_locker._compliance_state import explain_lock_decision
from screen_locker._status_projection import (
    _shutdown_projection,
    format_summary_line,
)
from screen_locker._status_types import (
    DayStatus,
    ManualWorkoutBudgetStatus,
    ShutdownProjection,
    ShutdownProjectionDay,
    SickBudgetStatus,
    StatusSnapshot,
    WeeklySummary,
)

__all__ = [
    "DayStatus",
    "ManualWorkoutBudgetStatus",
    "ShutdownProjection",
    "ShutdownProjectionDay",
    "SickBudgetStatus",
    "StatusSnapshot",
    "WeeklySummary",
    "format_summary_line",
    "gather_status",
]
from screen_locker._constants import (
    EARLY_BIRD_PENDING_FILE,
    EXTRA_BENEFITS_FILE,
    MANUAL_WORKOUT_BUDGET_PER_7_DAYS,
    MANUAL_WORKOUT_BUDGET_PER_30_DAYS,
    SCHEDULED_SKIPS_FILE,
    SHUTDOWN_BASE_FILE,
    SHUTDOWN_CONFIG_FILE,
    SICK_BUDGET_PER_7_DAYS,
    SICK_BUDGET_PER_30_DAYS,
    SICK_BUDGET_PER_90_DAYS,
)
from screen_locker._extra_benefits import (
    current_streak,
    has_extended_early_bird,
    weekly_shutdown_bonus_hours,
)
from screen_locker._log_io import load_workout_log
from screen_locker._sick_tracker import (
    count_in_window,
    is_budget_exhausted,
    load_history,
)
from screen_locker._wake_state import has_workout_skip_today
from screen_locker._weekly_check import (
    COUNTED_WORKOUT_TYPES,
    MANUAL_WORKOUT_TYPE,
    VERIFIED_WORKOUT_TYPES,
    WEEKLY_WORKOUT_MINIMUM,
    is_relaxed_day,
)

if TYPE_CHECKING:
    from screen_locker._sick_tracker import SickHistory

_DEFAULT_LOG_FILE = Path(__file__).resolve().parent / "workout_log.json"


def _day_status(day_date: date, entries: list[dict], sick_days: set[str]) -> DayStatus:
    """Build a :class:`DayStatus` for one calendar day from its entry list.

    ``day_count`` mirrors the weekly rule (each verified workout counts, all of
    a day's manual entries count once) so the week total sums to
    :func:`~screen_locker._weekly_check.count_weekly_workouts`.
    """
    iso = day_date.isoformat()
    label = day_date.strftime("%a %b %d")
    types = tuple(
        str(e.get("workout_data", {}).get("type", ""))
        for e in entries
        if isinstance(e, dict)
    )
    verified = sum(1 for t in types if t in VERIFIED_WORKOUT_TYPES)
    has_manual = any(t == MANUAL_WORKOUT_TYPE for t in types)
    sources = [
        str(e.get("workout_data", {}).get("source", ""))
        for e in entries
        if isinstance(e, dict)
    ]
    return DayStatus(
        date=iso,
        label=label,
        entry_types=tuple(t for t in types if t),
        source=" · ".join(s for s in sources if s),
        counted=any(t in COUNTED_WORKOUT_TYPES for t in types),
        day_count=verified + (1 if has_manual else 0),
        is_sick_day=iso in sick_days,
    )


def _week_days(
    log_file: Path, sick_history: SickHistory, *, today_local: datetime
) -> tuple[DayStatus, ...]:
    """Return this ISO week's days from Monday through today."""
    log_data = load_workout_log(log_file)
    sick_days = set(sick_history.sick_days)
    today_date = today_local.date()
    monday = today_date - timedelta(days=today_date.weekday())
    days: list[DayStatus] = []
    day = monday
    while day <= today_date:
        days.append(_day_status(day, log_data.get(day.isoformat(), []), sick_days))
        day += timedelta(days=1)
    return tuple(days)


def _weekly_summary(days: tuple[DayStatus, ...]) -> WeeklySummary:
    """Summarize a week's workout count against the weekly minimum."""
    counted = sum(d.day_count for d in days)
    return WeeklySummary(
        days=days,
        counted_count=counted,
        minimum=WEEKLY_WORKOUT_MINIMUM,
        remaining=max(0, WEEKLY_WORKOUT_MINIMUM - counted),
        extra=max(0, counted - WEEKLY_WORKOUT_MINIMUM),
    )


def _sick_budget_status(history: SickHistory, *, today: str) -> SickBudgetStatus:
    """Summarize rolling sick-day budget usage."""
    return SickBudgetStatus(
        used_7d=count_in_window(history, 7, today=today),
        budget_7d=SICK_BUDGET_PER_7_DAYS,
        used_30d=count_in_window(history, 30, today=today),
        budget_30d=SICK_BUDGET_PER_30_DAYS,
        used_90d=count_in_window(history, 90, today=today),
        budget_90d=SICK_BUDGET_PER_90_DAYS,
        debt=history.debt,
        exhausted=is_budget_exhausted(history, today=today),
    )


def _manual_workout_budget_status(
    log_file: Path, *, today: str
) -> ManualWorkoutBudgetStatus:
    """Summarize rolling manual-workout budget usage."""
    return ManualWorkoutBudgetStatus(
        used_7d=_manual_workout.count_in_window(log_file, 7, today=today),
        budget_7d=MANUAL_WORKOUT_BUDGET_PER_7_DAYS,
        used_30d=_manual_workout.count_in_window(log_file, 30, today=today),
        budget_30d=MANUAL_WORKOUT_BUDGET_PER_30_DAYS,
        exhausted=_manual_workout.is_budget_exhausted(log_file, today=today),
    )


def gather_status(
    *,
    log_file: Path = _DEFAULT_LOG_FILE,
    extra_benefits_file: Path = EXTRA_BENEFITS_FILE,
    shutdown_base_file: Path = SHUTDOWN_BASE_FILE,
    shutdown_config_file: Path = SHUTDOWN_CONFIG_FILE,
    scheduled_skips_file: Path = SCHEDULED_SKIPS_FILE,
    early_bird_pending_file: Path = EARLY_BIRD_PENDING_FILE,
    now: datetime | None = None,
) -> StatusSnapshot:
    """Gather a full :class:`StatusSnapshot` from on-disk state only.

    Never touches ADB, sudo, or the network — safe to call on every i3blocks
    tick and every status-window open/refresh.
    """
    instant = now if now is not None else datetime.now(tz=timezone.utc)
    today_local = instant.astimezone()
    today_str = today_local.date().isoformat()

    sick_history = load_history()
    days = _week_days(log_file, sick_history, today_local=today_local)
    week = _weekly_summary(days)
    early_bird_extended = has_extended_early_bird(
        extra_benefits_file, today=today_local
    )

    lock_explanation = explain_lock_decision(
        log_file=log_file,
        scheduled_skips_file=scheduled_skips_file,
        early_bird_pending_file=early_bird_pending_file,
        sick_history=sick_history,
        extended_early_bird=early_bird_extended,
        weekly_minimum_met=week.counted_count >= WEEKLY_WORKOUT_MINIMUM,
        relaxed_day=is_relaxed_day(today=today_local),
        wake_skip=has_workout_skip_today(),
        now=instant,
    )

    return StatusSnapshot(
        today=days[-1],
        week=week,
        bonus_hours_this_week=weekly_shutdown_bonus_hours(
            extra_benefits_file, today=today_local
        ),
        streak=current_streak(extra_benefits_file),
        early_bird_extended=early_bird_extended,
        shutdown=_shutdown_projection(
            shutdown_base_file=shutdown_base_file,
            shutdown_config_file=shutdown_config_file,
            extra_benefits_file=extra_benefits_file,
            log_file=log_file,
            today_local=today_local,
        ),
        lock_explanation=lock_explanation,
        sick_budget=_sick_budget_status(sick_history, today=today_str),
        manual_workout_budget=_manual_workout_budget_status(log_file, today=today_str),
        generated_at=instant.isoformat(),
    )