mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-11-10 19:07:05 +01:00
Show which games make up the gaming budget, and bill osu!lazer
The budget tab answered "how much did I play" but not "what did I play":
playtime_state.json and playtime_history.json each stored one aggregate
seconds figure per gaming day, so the 14-day chart was a row of featureless
bars. Separately, qualifying_pids had exactly two sources — the SteamAppId
env var and a hardcoded 16-launcher frozenset — so osu!lazer, which sets no
SteamAppId and matches no launcher name, was never billed at all.
qualifying_pids now returns pid -> attribution key instead of a bare pid set.
The app id was already in hand and thrown away on its first line; it now
survives into PlaytimeState.per_game, alongside a user-editable
counted_processes list in config.json seeded with osu!lazer.
Two things the parts/whole relationship forces:
- per_game only ever sums to AT MOST seconds, never equals it. backdate
refunds the idle grace on every engaged-to-paused edge, and
_state_or_recover synthesises a full budget with no breakdown. The
refund is debited from last_credited_key, and the remainder is rendered
as "Unattributed" rather than stored — one path then covers legacy days,
focus-probe failures and drift alike.
- counted_processes is kept apart from LAUNCHER_PROCESS_NAMES. That
frozenset is also the total-block kill list, and it routes through
kill_and_uninstall_launchers, which pacman -R's the owning package.
Counted entries are killed on total block, never uninstalled.
Both storage formats stay at schema 1: load_state already filters to
dataclass fields, so a record without per_game migrates for free, and the
history reader has to accept both a bare float and an object regardless, so
a version bump would distinguish nothing.
Chart segments use the shared design system's categorical ramp (--cat-1..6),
capped at six games plus "Other" because a seventh distinguishable hue does
not exist under simulated CVD. The red over-budget fill is gone: colour now
encodes which game, and the dashed budget line alone says over or under.
Snapshot persistence moves to _snapshot.py to keep config.py under the
250-line cap. Three modules bound config paths at import time, which
bypasses the conftest redirect — they now reach them through the module.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MKFzwzwRU1JD7SFLHtQBsz
This commit is contained in:
parent
2c12edacc4
commit
1faf211767
@ -187,7 +187,8 @@ repos:
|
||||
# pnpm-lock.yaml is integrity hashes, not prose: `vEw` in a
|
||||
# sha512 read as a misspelling of "view".
|
||||
- --skip=*.json,*.lock,*-lock.yaml,.git,__pycache__,.venv,node_modules
|
||||
- --ignore-words-list=als,ans,ect,nd,som,sur,te,nam,numer,lew,sie,wil,postion,clen,ther,folow,derrive,ony,tje,noe,theses,crate,doubleclick,wile,tabel,pary,blok,bloc,proces,serwer,parametr,adres,hart,dout,metod,tekst,synonim,grup,mosty,lokal,skalar,milion,nowe,tre,hel,alph
|
||||
# `lazer` is osu!lazer's own name, not a misspelling of laser.
|
||||
- --ignore-words-list=lazer,als,ans,ect,nd,som,sur,te,nam,numer,lew,sie,wil,postion,clen,ther,folow,derrive,ony,tje,noe,theses,crate,doubleclick,wile,tabel,pary,blok,bloc,proces,serwer,parametr,adres,hart,dout,metod,tekst,synonim,grup,mosty,lokal,skalar,milion,nowe,tre,hel,alph
|
||||
|
||||
# ===========================================================================
|
||||
# SHELLCHECK - Shell script linting
|
||||
|
||||
92
steam_backlog_enforcer/_attribution_labels.py
Normal file
92
steam_backlog_enforcer/_attribution_labels.py
Normal file
@ -0,0 +1,92 @@
|
||||
"""Turning attribution keys into names a human recognises.
|
||||
|
||||
Keys are stable (``app:475150``); labels are not (Steam renames games, and a
|
||||
user can retitle a ``counted_processes`` entry). So keys are what get stored in
|
||||
state and history, and this module resolves them for display.
|
||||
|
||||
Steam names come from ``appmanifest_<id>.acf`` rather than ``snapshot.json``:
|
||||
a game that is *running* is by definition installed, so its manifest exists and
|
||||
is a few hundred bytes, whereas the snapshot is ~11 MB and would be read on
|
||||
every three-second tick and every five-second UI poll.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
from steam_backlog_enforcer import _steam_state
|
||||
from steam_backlog_enforcer._counted_procs import labels_by_key, load_counted_processes
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterable
|
||||
|
||||
_APP_PREFIX: Final = "app:"
|
||||
_PROC_PREFIX: Final = "proc:"
|
||||
_LAUNCHER_PREFIX: Final = "launcher:"
|
||||
|
||||
# Same pattern game_uninstall.get_installed_games() uses on the same files.
|
||||
_NAME_RE: Final = re.compile(r'"name"\s+"([^"]+)"')
|
||||
|
||||
# Manifest names never change while a game runs, and the daemon is long-lived.
|
||||
_app_names: dict[int, str] = {}
|
||||
|
||||
|
||||
def _steam_name(app_id: int) -> str | None:
|
||||
"""Return *app_id*'s name from its appmanifest, or ``None``.
|
||||
|
||||
Args:
|
||||
app_id: Steam application id.
|
||||
|
||||
Returns:
|
||||
The manifest's ``name`` field, or ``None`` if unreadable.
|
||||
"""
|
||||
cached = _app_names.get(app_id)
|
||||
if cached is not None:
|
||||
return cached
|
||||
manifest = _steam_state.STEAMAPPS_PATH / f"appmanifest_{app_id}.acf"
|
||||
with contextlib.suppress(OSError):
|
||||
match = _NAME_RE.search(manifest.read_text(encoding="utf-8"))
|
||||
if match:
|
||||
name = match.group(1)
|
||||
_app_names[app_id] = name
|
||||
return name
|
||||
return None
|
||||
|
||||
|
||||
def label_for(key: str) -> str:
|
||||
"""Return a display name for an attribution *key*.
|
||||
|
||||
Falls back to the key itself, which is ugly but never wrong — an
|
||||
unresolvable game is still better identified by ``app:475150`` than by a
|
||||
blank row.
|
||||
|
||||
Args:
|
||||
key: An ``app:``/``proc:``/``launcher:`` attribution key.
|
||||
|
||||
Returns:
|
||||
The display label.
|
||||
"""
|
||||
if key.startswith(_APP_PREFIX):
|
||||
raw = key[len(_APP_PREFIX) :]
|
||||
if raw.isdigit():
|
||||
return _steam_name(int(raw)) or key
|
||||
return key
|
||||
if key.startswith(_PROC_PREFIX):
|
||||
return labels_by_key(load_counted_processes()).get(key, key)
|
||||
if key.startswith(_LAUNCHER_PREFIX):
|
||||
return key[len(_LAUNCHER_PREFIX) :]
|
||||
return key
|
||||
|
||||
|
||||
def labels_for(keys: Iterable[str]) -> dict[str, str]:
|
||||
"""Resolve every key in *keys* to its label.
|
||||
|
||||
Args:
|
||||
keys: Any iterable of attribution keys.
|
||||
|
||||
Returns:
|
||||
Mapping of key to display label.
|
||||
"""
|
||||
return {key: label_for(key) for key in keys}
|
||||
201
steam_backlog_enforcer/_budget_games.py
Normal file
201
steam_backlog_enforcer/_budget_games.py
Normal file
@ -0,0 +1,201 @@
|
||||
"""Rolling billed time up into the per-game slices the UI plots.
|
||||
|
||||
Two rules live here rather than in the chart component, so that the HTTP
|
||||
payload and the MCP payload cannot disagree with what is drawn:
|
||||
|
||||
**Unattributed is computed, never stored.** ``per_game`` only ever sums to at
|
||||
most the day's total — a tick the focus probe could not pin to one game still
|
||||
bills, ``backdate`` refunds scale the parts, and a fail-closed day has a full
|
||||
budget with no breakdown at all. The shortfall is surfaced as its own slice
|
||||
instead of being hidden or charged to a guess. Days written before per-game
|
||||
attribution existed have no breakdown, so their whole bar is this slice.
|
||||
|
||||
**The window is capped at six games plus "Other".** Six is not arbitrary: it is
|
||||
the size of the shared categorical ramp, which cannot be extended to seven
|
||||
distinguishable hues under simulated colour-vision deficiency.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from steam_backlog_enforcer._attribution_labels import label_for
|
||||
from steam_backlog_enforcer._playtime_history import load_labels
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from steam_backlog_enforcer._playtime_history import HistoryDay
|
||||
from steam_backlog_enforcer._playtime_state import PlaytimeState
|
||||
|
||||
UNATTRIBUTED_KEY: Final = "unattributed"
|
||||
OTHER_KEY: Final = "other"
|
||||
|
||||
# Matches the categorical ramp's six hues; see the module docstring.
|
||||
_TOP_N: Final = 6
|
||||
|
||||
|
||||
def _label(key: str, stored: dict[str, str]) -> str:
|
||||
"""Return a display label for *key*.
|
||||
|
||||
The stored map is preferred because it was written while the game was
|
||||
installed — an uninstalled game's appmanifest is gone, so a live lookup
|
||||
could not recover its name. The live lookup is the fallback for *today*,
|
||||
whose keys have not reached the history file yet.
|
||||
|
||||
Args:
|
||||
key: Attribution key.
|
||||
stored: Labels persisted alongside the history.
|
||||
|
||||
Returns:
|
||||
The display label.
|
||||
"""
|
||||
if key == UNATTRIBUTED_KEY:
|
||||
return "Unattributed"
|
||||
if key == OTHER_KEY:
|
||||
return "Other"
|
||||
return stored.get(key) or label_for(key)
|
||||
|
||||
|
||||
def _residual(total: float, games: dict[str, float]) -> float:
|
||||
"""Return the part of *total* that no key accounts for.
|
||||
|
||||
Floored at zero: proportional refunds make the parts drift by fractions of
|
||||
a second, and a negative segment would render as an inverted bar.
|
||||
|
||||
Args:
|
||||
total: Seconds billed that day.
|
||||
games: Attribution key to seconds.
|
||||
|
||||
Returns:
|
||||
The unattributed remainder.
|
||||
"""
|
||||
return max(0.0, total - sum(games.values()))
|
||||
|
||||
|
||||
def today_games(state: PlaytimeState) -> list[dict[str, Any]]:
|
||||
"""Return every game billed today, largest first, with the remainder last.
|
||||
|
||||
Uncapped on purpose: the fortnight chart has six colours to spend, but the
|
||||
"Today" list is text and can show everything.
|
||||
|
||||
Args:
|
||||
state: Loaded accounting state.
|
||||
|
||||
Returns:
|
||||
Serialisable slices, each with ``key``, ``label``, ``seconds`` and
|
||||
``fraction`` of the day's billed total.
|
||||
"""
|
||||
stored = load_labels()
|
||||
slices = sorted(state.per_game.items(), key=lambda item: -item[1])
|
||||
residual = _residual(state.seconds, state.per_game)
|
||||
if residual > 0:
|
||||
slices.append((UNATTRIBUTED_KEY, residual))
|
||||
|
||||
total = state.seconds or 1.0
|
||||
return [
|
||||
{
|
||||
"key": key,
|
||||
"label": _label(key, stored),
|
||||
"seconds": round(seconds, 1),
|
||||
"fraction": round(seconds / total, 4),
|
||||
}
|
||||
for key, seconds in slices
|
||||
if seconds > 0
|
||||
]
|
||||
|
||||
|
||||
def _window_top_keys(days: list[HistoryDay]) -> list[str]:
|
||||
"""Return the keys worth their own colour across the whole window.
|
||||
|
||||
Ranked by total across the window rather than per day, so a game keeps the
|
||||
same colour in every bar it appears in.
|
||||
|
||||
Args:
|
||||
days: The days being plotted.
|
||||
|
||||
Returns:
|
||||
At most ``_TOP_N`` keys, largest total first.
|
||||
"""
|
||||
totals: dict[str, float] = {}
|
||||
for day in days:
|
||||
for key, seconds in day.games.items():
|
||||
totals[key] = totals.get(key, 0.0) + seconds
|
||||
ranked = sorted(totals.items(), key=lambda item: (-item[1], item[0]))
|
||||
return [key for key, _ in ranked[:_TOP_N]]
|
||||
|
||||
|
||||
def _day_segments(day: HistoryDay, top: list[str]) -> dict[str, float]:
|
||||
"""Fold one day's breakdown into the capped key set.
|
||||
|
||||
Args:
|
||||
day: The day to fold.
|
||||
top: Keys that keep their own identity.
|
||||
|
||||
Returns:
|
||||
Mapping of rendered key to seconds.
|
||||
"""
|
||||
segments: dict[str, float] = {}
|
||||
for key, seconds in day.games.items():
|
||||
rendered = key if key in top else OTHER_KEY
|
||||
segments[rendered] = segments.get(rendered, 0.0) + seconds
|
||||
residual = _residual(day.seconds, day.games)
|
||||
if residual > 0:
|
||||
segments[UNATTRIBUTED_KEY] = segments.get(UNATTRIBUTED_KEY, 0.0) + residual
|
||||
return segments
|
||||
|
||||
|
||||
def history_view(
|
||||
days: list[HistoryDay],
|
||||
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
|
||||
"""Return the plottable days and the legend they share.
|
||||
|
||||
Segment order is the legend order, identical for every day, so a stacked
|
||||
bar's bands line up across the window and each key keeps one colour.
|
||||
|
||||
Args:
|
||||
days: Recorded days, oldest first.
|
||||
|
||||
Returns:
|
||||
The days and the legend, in render order.
|
||||
"""
|
||||
top = _window_top_keys(days)
|
||||
folded = [(day, _day_segments(day, top)) for day in days]
|
||||
|
||||
def present(key: str) -> bool:
|
||||
return any(key in seg for _, seg in folded)
|
||||
|
||||
# The two sentinels always trail the real games, in that order.
|
||||
order = [key for key in top if present(key)]
|
||||
order.extend(key for key in (OTHER_KEY, UNATTRIBUTED_KEY) if present(key))
|
||||
|
||||
stored = load_labels()
|
||||
legend = [{"key": key, "label": _label(key, stored)} for key in order]
|
||||
plotted = [
|
||||
{
|
||||
"day": day.day,
|
||||
"seconds": round(day.seconds, 1),
|
||||
"segments": [
|
||||
{"key": key, "seconds": round(segments[key], 1)}
|
||||
for key in order
|
||||
if segments.get(key, 0.0) > 0
|
||||
],
|
||||
}
|
||||
for day, segments in folded
|
||||
]
|
||||
return plotted, legend
|
||||
|
||||
|
||||
def billing_label(key: str) -> str:
|
||||
"""Return the display label for the key currently being credited.
|
||||
|
||||
Distinct from the backlog *assignment* the session block already reports:
|
||||
that is the game the enforcer told you to play, this is the one the budget
|
||||
is actually charging. They can legitimately differ — a counted non-Steam
|
||||
game is never the assignment.
|
||||
|
||||
Args:
|
||||
key: Attribution key, or ``""`` when nothing has billed yet.
|
||||
|
||||
Returns:
|
||||
The label, or ``""``.
|
||||
"""
|
||||
return _label(key, load_labels()) if key else ""
|
||||
@ -14,6 +14,11 @@ from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
from steam_backlog_enforcer._budget_games import (
|
||||
billing_label,
|
||||
history_view,
|
||||
today_games,
|
||||
)
|
||||
from steam_backlog_enforcer._budget_log_tail import last_verdict
|
||||
from steam_backlog_enforcer._playtime_block import mounted_targets
|
||||
from steam_backlog_enforcer._playtime_history import load_history
|
||||
@ -126,6 +131,7 @@ def build_today(state: PlaytimeState, rules: PlaytimeRules) -> dict[str, Any]:
|
||||
"blocked_at": state.blocked_at,
|
||||
"next_warning_seconds": next_warning(state, rules),
|
||||
"warned_seconds": list(state.warned_seconds),
|
||||
"games": today_games(state),
|
||||
}
|
||||
|
||||
|
||||
@ -172,6 +178,12 @@ def build_budget_snapshot(*, demo: bool = False) -> dict[str, Any]:
|
||||
# implying an empty day.
|
||||
status = access if access != READABLE else (READABLE if stored else CORRUPT)
|
||||
session = last_verdict(demo=demo)
|
||||
# Demo runs have no history by design — HistoryWriter skips them, for the
|
||||
# same reason they get their own state file and log. Serving the production
|
||||
# days here would plot real 8-hour days against the demo's 60-second budget
|
||||
# line, painting every one of them over-budget. `today.games` is still
|
||||
# emitted: the demo's own breakdown is real.
|
||||
history, legend = ([], []) if demo else history_view(load_history(_HISTORY_DAYS))
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
@ -180,6 +192,9 @@ def build_budget_snapshot(*, demo: bool = False) -> dict[str, Any]:
|
||||
"error": _STATUS_MESSAGES.get(status),
|
||||
"today": build_today(stored, rules) if stored is not None else None,
|
||||
"session": {
|
||||
# What the budget is actually charging — a different question from
|
||||
# the backlog assignment reported as "game_name" below.
|
||||
"billing_label": billing_label(stored.last_credited_key if stored else ""),
|
||||
"available": session.available,
|
||||
"observed_at": session.observed_at,
|
||||
"state": session.state,
|
||||
@ -201,11 +216,7 @@ def build_budget_snapshot(*, demo: bool = False) -> dict[str, Any]:
|
||||
# the same reason they get their own state file and log. Serving the
|
||||
# production days here would plot real 8-hour days against the demo's
|
||||
# 60-second budget line, painting every one of them over-budget.
|
||||
"history": []
|
||||
if demo
|
||||
else [
|
||||
{"day": day.day, "seconds": round(day.seconds, 1)}
|
||||
for day in load_history(_HISTORY_DAYS)
|
||||
],
|
||||
"history": history,
|
||||
"legend": legend,
|
||||
"rules": build_rules(rules),
|
||||
}
|
||||
|
||||
@ -7,7 +7,7 @@ import logging
|
||||
import sys
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer.config import State, load_snapshot
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer.game_install import (
|
||||
_echo,
|
||||
)
|
||||
@ -20,6 +20,7 @@ from steam_backlog_enforcer.hltb import (
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from steam_backlog_enforcer.config import State
|
||||
from steam_backlog_enforcer.steam_api import GameInfo
|
||||
|
||||
_REASSIGN_REFRESH_LIMIT = 50
|
||||
|
||||
@ -7,6 +7,7 @@ under the 250-line cap.
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from steam_backlog_enforcer._actions import allowed_app_ids
|
||||
from steam_backlog_enforcer._cmd_done import (
|
||||
@ -17,7 +18,7 @@ from steam_backlog_enforcer._cmd_done import (
|
||||
)
|
||||
from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
|
||||
from steam_backlog_enforcer._pick_completion import mark_finished
|
||||
from steam_backlog_enforcer.config import Config, State, load_snapshot
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer.enforcer import (
|
||||
enforce_allowed_game,
|
||||
send_notification,
|
||||
@ -36,6 +37,9 @@ from steam_backlog_enforcer.library_hider import try_hide_other_games
|
||||
from steam_backlog_enforcer.scanning import pick_next_game
|
||||
from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
|
||||
_REASSIGN_REFRESH_LIMIT = 50
|
||||
_SKIP_DAYS = 7
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
245
steam_backlog_enforcer/_counted_procs.py
Normal file
245
steam_backlog_enforcer/_counted_procs.py
Normal file
@ -0,0 +1,245 @@
|
||||
"""User-declared non-Steam processes that count against the gaming budget.
|
||||
|
||||
Steam games announce themselves through ``SteamAppId``; everything else has to
|
||||
be named. :mod:`_total_block_launchers` already names sixteen *launchers*, but
|
||||
that frozenset is consumed by the total-block killer, so adding a game to it
|
||||
silently makes the game killable. This module is the separate, user-editable
|
||||
list: ``counted_processes`` in ``config.json``.
|
||||
|
||||
Loading is deliberately independent of :class:`~config.Config`. The two kill
|
||||
paths that need these names — ``_playtime_kill.steam_and_launcher_pids`` and
|
||||
``_total_block.enforce_total_block_tick`` — take no arguments and never build a
|
||||
``Config``, so a ``Config``-only accessor could not reach them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from steam_backlog_enforcer import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Names that must never be matched. Every name here is either an interpreter
|
||||
# that hosts unrelated programs or a shell: `counted_processes` entries are
|
||||
# kill targets, and `sh` in the set would SIGTERM every shell on the system.
|
||||
# osu!lazer is launched through a `/bin/sh` wrapper, so this is a live hazard,
|
||||
# not a hypothetical one.
|
||||
FORBIDDEN_NAMES = frozenset(
|
||||
{
|
||||
"bash",
|
||||
"dash",
|
||||
"dotnet",
|
||||
"env",
|
||||
"java",
|
||||
"mono",
|
||||
"node",
|
||||
"perl",
|
||||
"python",
|
||||
"python3",
|
||||
"ruby",
|
||||
"sh",
|
||||
"wine",
|
||||
"wine64",
|
||||
"zsh",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CountedProcess:
|
||||
"""One user-declared non-Steam game that bills the budget."""
|
||||
|
||||
id: str
|
||||
"""Stable slug; forms the ``proc:<id>`` attribution key."""
|
||||
label: str
|
||||
"""Display name shown in the UI and stored in history labels."""
|
||||
names: frozenset[str]
|
||||
"""Process basenames to match against ``comm`` and argv."""
|
||||
|
||||
|
||||
# osu!lazer ships as the AUR `osu-lazer-bin` package: `/usr/bin/osu-lazer` is a
|
||||
# `/bin/sh` wrapper that execs `/opt/osu-lazer/osu.AppImage`, which self-mounts
|
||||
# a squashfs and execs the game itself. All three layers are named so that
|
||||
# whichever one the detector sees first, the time is billed.
|
||||
DEFAULT_COUNTED_PROCESSES: tuple[CountedProcess, ...] = (
|
||||
CountedProcess(
|
||||
id="osu-lazer",
|
||||
label="osu!lazer",
|
||||
names=frozenset({"osu-lazer", "osu.AppImage", "osu!"}),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def parse_counted_processes(raw: object) -> tuple[CountedProcess, ...]:
|
||||
"""Build validated entries from ``config.json``'s ``counted_processes``.
|
||||
|
||||
Invalid entries are dropped with a warning rather than raising: a typo in a
|
||||
hand-edited config must not take the whole enforcer down, and the failure
|
||||
mode of dropping one entry (that game stops billing) is visible in the UI.
|
||||
|
||||
Args:
|
||||
raw: The deserialised ``counted_processes`` value.
|
||||
|
||||
Returns:
|
||||
The valid entries, in file order.
|
||||
"""
|
||||
if not isinstance(raw, list):
|
||||
logger.warning("counted_processes is not a list; ignoring it.")
|
||||
return ()
|
||||
|
||||
entries: list[CountedProcess] = []
|
||||
seen: set[str] = set()
|
||||
for item in raw:
|
||||
entry = _parse_entry(item)
|
||||
if entry is None or entry.id in seen:
|
||||
continue
|
||||
seen.add(entry.id)
|
||||
entries.append(entry)
|
||||
return tuple(entries)
|
||||
|
||||
|
||||
def _parse_entry(item: object) -> CountedProcess | None:
|
||||
"""Validate one ``counted_processes`` element.
|
||||
|
||||
Args:
|
||||
item: A single deserialised entry.
|
||||
|
||||
Returns:
|
||||
The entry, or ``None`` if it is unusable.
|
||||
"""
|
||||
if not isinstance(item, dict):
|
||||
logger.warning("counted_processes entry is not an object; skipping.")
|
||||
return None
|
||||
|
||||
entry_id = _text(item.get("id"))
|
||||
if not entry_id:
|
||||
logger.warning("counted_processes entry has no id; skipping.")
|
||||
return None
|
||||
|
||||
names = _clean_names(entry_id, item.get("names"))
|
||||
if not names:
|
||||
logger.warning("counted_processes entry %r has no usable names.", entry_id)
|
||||
return None
|
||||
|
||||
return CountedProcess(
|
||||
id=entry_id,
|
||||
label=_text(item.get("label")) or entry_id,
|
||||
names=names,
|
||||
)
|
||||
|
||||
|
||||
def _clean_names(entry_id: str, raw: object) -> frozenset[str]:
|
||||
"""Return the matchable names in *raw*, dropping forbidden ones.
|
||||
|
||||
Args:
|
||||
entry_id: Owning entry id, for the warning message.
|
||||
raw: The deserialised ``names`` value.
|
||||
|
||||
Returns:
|
||||
The usable names.
|
||||
"""
|
||||
if not isinstance(raw, list):
|
||||
return frozenset()
|
||||
|
||||
names: set[str] = set()
|
||||
for value in raw:
|
||||
name = _text(value)
|
||||
if not name:
|
||||
continue
|
||||
if name in FORBIDDEN_NAMES:
|
||||
logger.warning(
|
||||
"counted_processes entry %r names %r, which hosts unrelated "
|
||||
"programs and is a kill target; refusing it.",
|
||||
entry_id,
|
||||
name,
|
||||
)
|
||||
continue
|
||||
names.add(name)
|
||||
return frozenset(names)
|
||||
|
||||
|
||||
def _text(value: object) -> str:
|
||||
"""Return *value* as a stripped string, or ``""`` if it is not one.
|
||||
|
||||
Args:
|
||||
value: Any deserialised JSON value.
|
||||
|
||||
Returns:
|
||||
The trimmed string.
|
||||
"""
|
||||
return value.strip() if isinstance(value, str) else ""
|
||||
|
||||
|
||||
def load_counted_processes() -> tuple[CountedProcess, ...]:
|
||||
"""Read ``counted_processes`` straight from ``config.json``.
|
||||
|
||||
Used by the kill paths, which have no ``Config`` to consult. A missing or
|
||||
unreadable file yields the defaults, so a fresh install still bills
|
||||
osu!lazer.
|
||||
|
||||
Returns:
|
||||
The configured entries, or :data:`DEFAULT_COUNTED_PROCESSES`.
|
||||
"""
|
||||
try:
|
||||
data: Any = json.loads(config.CONFIG_FILE.read_text(encoding="utf-8"))
|
||||
except OSError, ValueError:
|
||||
return DEFAULT_COUNTED_PROCESSES
|
||||
|
||||
if not isinstance(data, dict) or "counted_processes" not in data:
|
||||
return DEFAULT_COUNTED_PROCESSES
|
||||
return parse_counted_processes(data["counted_processes"])
|
||||
|
||||
|
||||
def counted_process_names(entries: tuple[CountedProcess, ...]) -> frozenset[str]:
|
||||
"""Return every matchable name across *entries*.
|
||||
|
||||
Args:
|
||||
entries: Configured entries.
|
||||
|
||||
Returns:
|
||||
The union of their names.
|
||||
"""
|
||||
if not entries:
|
||||
return frozenset()
|
||||
return frozenset().union(*(entry.names for entry in entries))
|
||||
|
||||
|
||||
def key_by_name(entries: tuple[CountedProcess, ...]) -> dict[str, str]:
|
||||
"""Map each matchable name to its ``proc:<id>`` attribution key.
|
||||
|
||||
Args:
|
||||
entries: Configured entries.
|
||||
|
||||
Returns:
|
||||
Mapping of process name to attribution key.
|
||||
"""
|
||||
return {name: f"proc:{entry.id}" for entry in entries for name in entry.names}
|
||||
|
||||
|
||||
def labels_by_key(entries: tuple[CountedProcess, ...]) -> dict[str, str]:
|
||||
"""Map each ``proc:<id>`` key to its display label.
|
||||
|
||||
Args:
|
||||
entries: Configured entries.
|
||||
|
||||
Returns:
|
||||
Mapping of attribution key to label.
|
||||
"""
|
||||
return {f"proc:{entry.id}": entry.label for entry in entries}
|
||||
|
||||
|
||||
def kill_target_names() -> frozenset[str]:
|
||||
"""Return the counted-process names the kill paths should also target.
|
||||
|
||||
The budget cutoff and the total block kill these; the
|
||||
``kill_unauthorized_games`` path does not — it matches on ``SteamAppId``,
|
||||
which a non-Steam game never sets, so it excludes them for free.
|
||||
|
||||
Returns:
|
||||
Every configured matchable name.
|
||||
"""
|
||||
return counted_process_names(load_counted_processes())
|
||||
@ -24,7 +24,11 @@ from steam_backlog_enforcer._engagement_types import (
|
||||
PROBE_SCREEN_HELD,
|
||||
CauseTally,
|
||||
)
|
||||
from steam_backlog_enforcer._playtime_kill import _MAX_PROCESS_TREE_DEPTH, _read_ppid
|
||||
from steam_backlog_enforcer._playtime_kill import (
|
||||
_INIT_PID,
|
||||
_MAX_PROCESS_TREE_DEPTH,
|
||||
_read_ppid,
|
||||
)
|
||||
from steam_backlog_enforcer._screen_hold import ScreenHoldError, screen_hold
|
||||
from steam_backlog_enforcer._x_focus import focused_pid
|
||||
from steam_backlog_enforcer._x_probe import XProbeError
|
||||
@ -38,8 +42,6 @@ if TYPE_CHECKING:
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_INIT_PID = 1
|
||||
|
||||
|
||||
def assess_screen(
|
||||
holder_path: Path, tally: CauseTally
|
||||
|
||||
@ -61,6 +61,9 @@ def get_gaming_time() -> dict[str, Any]:
|
||||
"enforcement": rules["enforcement"],
|
||||
"counts_launchers": rules["counts_launchers"],
|
||||
"masked_launchers": masked,
|
||||
# Which games make up seconds_used, largest first, with whatever the
|
||||
# focus probe could not attribute as a trailing "unattributed" slice.
|
||||
"games": today["games"],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -17,8 +17,9 @@ from steam_backlog_enforcer._mcp_server import (
|
||||
_backlog_sort_key,
|
||||
mcp,
|
||||
)
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer._web_dataset import build_web_dataset, dataset_to_payload
|
||||
from steam_backlog_enforcer.config import State, load_snapshot
|
||||
from steam_backlog_enforcer.config import State
|
||||
from steam_backlog_enforcer.steam_api import GameInfo
|
||||
|
||||
|
||||
|
||||
@ -31,7 +31,7 @@ from typing import TYPE_CHECKING
|
||||
|
||||
from mcp.server import MCPServer
|
||||
|
||||
from steam_backlog_enforcer.config import load_snapshot
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from steam_backlog_enforcer.steam_api import GameInfo
|
||||
|
||||
@ -13,12 +13,8 @@ import logging
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from steam_backlog_enforcer.config import (
|
||||
CONFIG_DIR,
|
||||
Config,
|
||||
_atomic_write,
|
||||
load_snapshot,
|
||||
)
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer.config import CONFIG_DIR, Config, _atomic_write
|
||||
from steam_backlog_enforcer.steam_api import SteamAPIClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -46,7 +46,7 @@ from steam_backlog_enforcer._playtime_cutoff import (
|
||||
_sustain_block,
|
||||
_warn,
|
||||
)
|
||||
from steam_backlog_enforcer._playtime_procs import qualifying_pids
|
||||
from steam_backlog_enforcer._playtime_procs import attributed_key, qualifying_pids
|
||||
from steam_backlog_enforcer._playtime_state import (
|
||||
PlaytimeRules,
|
||||
PlaytimeState,
|
||||
@ -104,7 +104,8 @@ def playtime_tick(
|
||||
session.history.observe(state, demo=demo)
|
||||
return
|
||||
|
||||
pids = qualifying_pids(rules)
|
||||
owners = qualifying_pids(rules)
|
||||
pids = set(owners)
|
||||
monotonic = time.monotonic()
|
||||
verdict = session.tracker.assess(rules, qualifying=pids, now_monotonic=monotonic)
|
||||
|
||||
@ -113,6 +114,9 @@ def playtime_tick(
|
||||
now=now,
|
||||
qualifying=pids if verdict.engaged else set(),
|
||||
interval=interval,
|
||||
credited_key=(
|
||||
attributed_key(owners, verdict.focus_pid) if verdict.engaged else ""
|
||||
),
|
||||
)
|
||||
state = session.tracker.backdate(state, verdict, rules=rules)
|
||||
state = _policy(state, rules, now=now)
|
||||
|
||||
@ -29,6 +29,7 @@ def accumulate(
|
||||
now: datetime,
|
||||
qualifying: set[int],
|
||||
interval: float,
|
||||
credited_key: str = "",
|
||||
) -> PlaytimeState:
|
||||
"""Add this tick's elapsed gaming time to *state*.
|
||||
|
||||
@ -44,6 +45,11 @@ def accumulate(
|
||||
now: Timezone-aware local timestamp for this tick.
|
||||
qualifying: PIDs currently counting against the budget.
|
||||
interval: Nominal seconds between ticks.
|
||||
credited_key: Attribution key to charge, or ``""`` when the tick cannot
|
||||
be pinned to one game. An unattributable tick still bills
|
||||
``seconds``; the shortfall is rendered as Unattributed rather than
|
||||
charged to a guess, which is why ``per_game`` only ever sums to at
|
||||
most ``seconds``.
|
||||
|
||||
Returns:
|
||||
The updated state.
|
||||
@ -57,7 +63,16 @@ def accumulate(
|
||||
return replace(state, last_tick_at=stamp)
|
||||
|
||||
credited = min(delta, interval * _DELTA_CLAMP_FACTOR)
|
||||
return replace(state, last_tick_at=stamp, seconds=state.seconds + credited)
|
||||
per_game = dict(state.per_game)
|
||||
if credited_key:
|
||||
per_game[credited_key] = per_game.get(credited_key, 0.0) + credited
|
||||
return replace(
|
||||
state,
|
||||
last_tick_at=stamp,
|
||||
seconds=state.seconds + credited,
|
||||
per_game=per_game,
|
||||
last_credited_key=credited_key or state.last_credited_key,
|
||||
)
|
||||
|
||||
|
||||
def roll_over(state: PlaytimeState, *, day_key: str) -> PlaytimeState:
|
||||
|
||||
@ -122,4 +122,4 @@ def _kill_set(rules: PlaytimeRules) -> set[int]:
|
||||
Returns:
|
||||
PIDs to signal.
|
||||
"""
|
||||
return qualifying_pids(rules) | steam_and_launcher_pids()
|
||||
return set(qualifying_pids(rules)) | steam_and_launcher_pids()
|
||||
|
||||
@ -156,9 +156,17 @@ class EngagementTracker:
|
||||
return state
|
||||
|
||||
refunded = max(0.0, state.seconds - rules.idle_grace_seconds)
|
||||
amount = state.seconds - refunded
|
||||
logger.info(
|
||||
"Idle for %.0fs; refunding the %.0fs grace period.",
|
||||
verdict.idle_seconds or 0.0,
|
||||
state.seconds - refunded,
|
||||
amount,
|
||||
)
|
||||
return replace(state, seconds=refunded)
|
||||
# The refund must come off the same key it was billed to, or the
|
||||
# per-game parts drift above the whole on every walk-away — which is
|
||||
# many times a day, and would render as segments overflowing the bar.
|
||||
per_game = dict(state.per_game)
|
||||
key = state.last_credited_key
|
||||
if key and key in per_game:
|
||||
per_game[key] = max(0.0, per_game[key] - amount)
|
||||
return replace(state, seconds=refunded, per_game=per_game)
|
||||
|
||||
@ -7,8 +7,10 @@ five 5 MB files — so a busy fortnight can push a day out of the window entirel
|
||||
Neither is a series you can plot.
|
||||
|
||||
So this module keeps one small file whose only job is the history: a
|
||||
``day_key -> seconds`` mapping, trimmed to the last month, written world-readable
|
||||
so an unprivileged reader (the web UI) can plot it.
|
||||
``day_key -> {seconds, games}`` mapping, trimmed to the last month, written
|
||||
world-readable so an unprivileged reader (the web UI) can plot it. Schema 1
|
||||
stored a bare seconds float per day and is still read: those days simply carry
|
||||
no per-game breakdown, which the payload renders as Unattributed.
|
||||
|
||||
Writes are throttled rather than tied to roll-over. Upserting today's figure on a
|
||||
cadence means the outgoing day is already recorded to within ``_FLUSH_DELTA``
|
||||
@ -22,11 +24,16 @@ failure and must never take down the enforcer.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import json
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
from steam_backlog_enforcer._attribution_labels import labels_for
|
||||
from steam_backlog_enforcer._playtime_history_parse import (
|
||||
HistoryDay,
|
||||
_parse_day,
|
||||
_parse_labels,
|
||||
)
|
||||
from steam_backlog_enforcer._playtime_state import STATE_MODE
|
||||
from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
|
||||
|
||||
@ -35,10 +42,17 @@ if TYPE_CHECKING:
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = ["HISTORY_FILE", "HistoryDay", "HistoryWriter", "load_history", "record_day"]
|
||||
|
||||
HISTORY_FILE: Final = CONFIG_DIR / "playtime_history.json"
|
||||
|
||||
_SCHEMA_VERSION: Final = 1
|
||||
|
||||
# Day values are polymorphic rather than versioned: a bare float is the
|
||||
# original `day -> seconds` form, an object carries the per-game breakdown.
|
||||
# The reader must handle both either way, so a version bump would add a
|
||||
# branch that distinguishes nothing.
|
||||
|
||||
# Days retained. Comfortably more than the fortnight the UI plots, so a chart
|
||||
# that later grows a longer window has data waiting for it.
|
||||
_MAX_DAYS: Final = 30
|
||||
@ -49,18 +63,8 @@ _MAX_DAYS: Final = 30
|
||||
_FLUSH_DELTA: Final = 60.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HistoryDay:
|
||||
"""One gaming day's total, as plotted."""
|
||||
|
||||
day: str
|
||||
"""Gaming day, ``YYYY-MM-DD``, 06:00-shifted."""
|
||||
seconds: float
|
||||
"""Qualifying seconds billed during ``day``."""
|
||||
|
||||
|
||||
def _read_days() -> dict[str, float]:
|
||||
"""Return the stored ``day -> seconds`` mapping, or empty if unusable.
|
||||
def _read_file() -> tuple[dict[str, HistoryDay], dict[str, str]]:
|
||||
"""Return the stored days and label map, or empties if unusable.
|
||||
|
||||
Unlike the state file this one is neither root-owned nor immutable, and it
|
||||
lives in a user-owned directory — so an unprivileged process can replace it
|
||||
@ -68,29 +72,34 @@ def _read_days() -> dict[str, float]:
|
||||
field is therefore checked rather than trusted.
|
||||
|
||||
Returns:
|
||||
The mapping, or ``{}`` when the file is missing, unreadable, corrupt or
|
||||
of an unknown schema.
|
||||
The ``day -> HistoryDay`` mapping and the ``key -> label`` mapping, both
|
||||
empty when the file is missing, unreadable, corrupt or of an unknown
|
||||
schema.
|
||||
"""
|
||||
try:
|
||||
raw = json.loads(HISTORY_FILE.read_text(encoding="utf-8"))
|
||||
except FileNotFoundError:
|
||||
return {}
|
||||
return {}, {}
|
||||
except OSError, ValueError:
|
||||
logger.warning(
|
||||
"Playtime history at %s is unreadable; starting over.", HISTORY_FILE
|
||||
)
|
||||
return {}
|
||||
return {}, {}
|
||||
if not isinstance(raw, dict) or raw.get("schema_version") != _SCHEMA_VERSION:
|
||||
logger.warning("Playtime history at %s has an unknown schema.", HISTORY_FILE)
|
||||
return {}
|
||||
return {}, {}
|
||||
|
||||
days = raw.get("days")
|
||||
if not isinstance(days, dict):
|
||||
return {}
|
||||
return {
|
||||
key: float(value)
|
||||
return {}, {}
|
||||
parsed = {
|
||||
key: entry
|
||||
for key, value in days.items()
|
||||
if isinstance(key, str) and isinstance(value, (int, float))
|
||||
if isinstance(key, str)
|
||||
for entry in [_parse_day(key, value)]
|
||||
if entry is not None
|
||||
}
|
||||
return parsed, _parse_labels(raw.get("labels"))
|
||||
|
||||
|
||||
def load_history(limit: int = 14) -> list[HistoryDay]:
|
||||
@ -102,25 +111,50 @@ def load_history(limit: int = 14) -> list[HistoryDay]:
|
||||
Returns:
|
||||
Recorded days in ascending date order, at most *limit* of them.
|
||||
"""
|
||||
days = _read_days()
|
||||
days, _ = _read_file()
|
||||
ordered = sorted(days.items())[-limit:] if limit > 0 else []
|
||||
return [HistoryDay(day=day, seconds=seconds) for day, seconds in ordered]
|
||||
return [entry for _, entry in ordered]
|
||||
|
||||
|
||||
def record_day(day_key: str, seconds: float) -> None:
|
||||
"""Upsert one day's total and trim the file to ``_MAX_DAYS``.
|
||||
def load_labels() -> dict[str, str]:
|
||||
"""Return the stored ``key -> label`` mapping.
|
||||
|
||||
Returns:
|
||||
Labels recorded alongside the history.
|
||||
"""
|
||||
return _read_file()[1]
|
||||
|
||||
|
||||
def record_day(day_key: str, seconds: float, games: dict[str, float]) -> None:
|
||||
"""Upsert one day's total and breakdown, trimming to ``_MAX_DAYS``.
|
||||
|
||||
Labels are merged rather than replaced so a game that stops being played
|
||||
keeps its name for as long as its days remain in the window.
|
||||
|
||||
Args:
|
||||
day_key: Gaming day to record.
|
||||
seconds: Total billed for that day so far.
|
||||
games: Attribution key to seconds for that day.
|
||||
"""
|
||||
days = _read_days()
|
||||
days[day_key] = seconds
|
||||
days, labels = _read_file()
|
||||
days[day_key] = HistoryDay(day=day_key, seconds=seconds, games=dict(games))
|
||||
trimmed = dict(sorted(days.items())[-_MAX_DAYS:])
|
||||
labels.update(labels_for(games))
|
||||
payload = {
|
||||
"schema_version": _SCHEMA_VERSION,
|
||||
"days": {
|
||||
day: {"seconds": entry.seconds, "games": entry.games}
|
||||
for day, entry in trimmed.items()
|
||||
},
|
||||
"labels": {
|
||||
key: label
|
||||
for key, label in labels.items()
|
||||
if any(key in entry.games for entry in trimmed.values())
|
||||
},
|
||||
}
|
||||
_atomic_write(
|
||||
HISTORY_FILE,
|
||||
json.dumps({"schema_version": _SCHEMA_VERSION, "days": trimmed}, indent=2)
|
||||
+ "\n",
|
||||
json.dumps(payload, indent=2) + "\n",
|
||||
mode=STATE_MODE,
|
||||
)
|
||||
|
||||
@ -153,7 +187,7 @@ class HistoryWriter:
|
||||
if previous is not None and abs(state.seconds - previous) < _FLUSH_DELTA:
|
||||
return
|
||||
try:
|
||||
record_day(state.day_key, state.seconds)
|
||||
record_day(state.day_key, state.seconds, state.per_game)
|
||||
except OSError as exc:
|
||||
# A history point is cosmetic; the enforcer keeps enforcing.
|
||||
logger.warning("Could not record playtime history: %s", exc)
|
||||
|
||||
90
steam_backlog_enforcer/_playtime_history_parse.py
Normal file
90
steam_backlog_enforcer/_playtime_history_parse.py
Normal file
@ -0,0 +1,90 @@
|
||||
"""Parsing the stored playtime-history values.
|
||||
|
||||
Split from :mod:`steam_backlog_enforcer._playtime_history` to keep both modules
|
||||
under the 250-line cap. Everything here is pure: it turns already-deserialised
|
||||
JSON into validated values and touches neither the clock nor the disk.
|
||||
|
||||
The history file is user-owned and neither root-owned nor immutable, so an
|
||||
unprivileged process can replace it with anything at all and the web server
|
||||
will parse whatever it finds. Every field is therefore checked rather than
|
||||
trusted.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HistoryDay:
|
||||
"""One gaming day's total, as plotted."""
|
||||
|
||||
day: str
|
||||
"""Gaming day, ``YYYY-MM-DD``, 06:00-shifted."""
|
||||
seconds: float
|
||||
"""Qualifying seconds billed during ``day``."""
|
||||
games: dict[str, float] = field(default_factory=dict)
|
||||
"""Attribution key to seconds. Sums to at most ``seconds``; see below.
|
||||
|
||||
Empty for days written under schema 1, and short of ``seconds`` whenever a
|
||||
tick could not be attributed. The shortfall is rendered rather than stored.
|
||||
"""
|
||||
|
||||
|
||||
def _parse_day(day: str, value: object) -> HistoryDay | None:
|
||||
"""Build a :class:`HistoryDay` from one stored day value.
|
||||
|
||||
Args:
|
||||
day: The gaming day key.
|
||||
value: Either a schema-1 bare float or a schema-2 object.
|
||||
|
||||
Returns:
|
||||
The parsed day, or ``None`` if the value is not usable.
|
||||
"""
|
||||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||||
return HistoryDay(day=day, seconds=float(value))
|
||||
if not isinstance(value, dict):
|
||||
return None
|
||||
seconds = value.get("seconds")
|
||||
if not isinstance(seconds, (int, float)) or isinstance(seconds, bool):
|
||||
return None
|
||||
return HistoryDay(day=day, seconds=float(seconds), games=_parse_games(value))
|
||||
|
||||
|
||||
def _parse_games(value: dict[str, object]) -> dict[str, float]:
|
||||
"""Return the validated ``games`` mapping of a stored day.
|
||||
|
||||
Args:
|
||||
value: A schema-2 day object.
|
||||
|
||||
Returns:
|
||||
Attribution key to seconds, dropping anything malformed.
|
||||
"""
|
||||
games = value.get("games")
|
||||
if not isinstance(games, dict):
|
||||
return {}
|
||||
return {
|
||||
key: float(seconds)
|
||||
for key, seconds in games.items()
|
||||
if isinstance(key, str)
|
||||
and isinstance(seconds, (int, float))
|
||||
and not isinstance(seconds, bool)
|
||||
}
|
||||
|
||||
|
||||
def _parse_labels(value: object) -> dict[str, str]:
|
||||
"""Return the validated ``key -> label`` mapping.
|
||||
|
||||
Args:
|
||||
value: The stored ``labels`` value.
|
||||
|
||||
Returns:
|
||||
The mapping, dropping anything malformed.
|
||||
"""
|
||||
if not isinstance(value, dict):
|
||||
return {}
|
||||
return {
|
||||
key: label
|
||||
for key, label in value.items()
|
||||
if isinstance(key, str) and isinstance(label, str)
|
||||
}
|
||||
@ -15,6 +15,7 @@ from pathlib import Path
|
||||
import shutil
|
||||
import signal
|
||||
|
||||
from steam_backlog_enforcer._counted_procs import kill_target_names
|
||||
from steam_backlog_enforcer._playtime_run import _run
|
||||
from steam_backlog_enforcer._total_block_launchers import (
|
||||
LAUNCHER_PROCESS_NAMES,
|
||||
@ -29,6 +30,9 @@ _PROC = Path("/proc")
|
||||
_STAT_PPID_INDEX = 1
|
||||
_MAX_PROCESS_TREE_DEPTH = 32
|
||||
|
||||
# Walking past pid 1 is walking off the top of the process tree.
|
||||
_INIT_PID = 1
|
||||
|
||||
|
||||
def request_steam_shutdown() -> None:
|
||||
"""Ask Steam to close cleanly, without waiting for it.
|
||||
@ -50,7 +54,7 @@ def steam_and_launcher_pids() -> set[int]:
|
||||
Returns:
|
||||
The launcher and client PIDs.
|
||||
"""
|
||||
names = STEAM_CLIENT_PROCESS_NAMES | LAUNCHER_PROCESS_NAMES
|
||||
names = STEAM_CLIENT_PROCESS_NAMES | LAUNCHER_PROCESS_NAMES | kill_target_names()
|
||||
return set(get_pids_by_process_names(names))
|
||||
|
||||
|
||||
|
||||
@ -11,6 +11,12 @@ import os
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from steam_backlog_enforcer._counted_procs import key_by_name, load_counted_processes
|
||||
from steam_backlog_enforcer._playtime_kill import (
|
||||
_INIT_PID,
|
||||
_MAX_PROCESS_TREE_DEPTH,
|
||||
_read_ppid,
|
||||
)
|
||||
from steam_backlog_enforcer._total_block_launchers import LAUNCHER_PROCESS_NAMES
|
||||
from steam_backlog_enforcer.enforcer import (
|
||||
get_pids_by_process_names,
|
||||
@ -110,21 +116,99 @@ def process_name(pid: int) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def qualifying_pids(rules: PlaytimeRules) -> set[int]:
|
||||
"""Return PIDs whose runtime counts against the daily budget.
|
||||
def _merge_named(found: dict[int, str], keys: dict[str, str]) -> None:
|
||||
"""Add name-matched PIDs to *found* without displacing existing keys.
|
||||
|
||||
``setdefault`` is the point: a Steam game already attributed by its
|
||||
``SteamAppId`` must keep that key even if its ``comm`` also happens to match
|
||||
a launcher name.
|
||||
|
||||
A scanner can only return names it was asked for, so an unmapped name is
|
||||
unreachable in production — it is skipped rather than raising, because the
|
||||
alternative is a ``KeyError`` inside the enforce loop.
|
||||
|
||||
Args:
|
||||
found: Accumulating PID-to-key mapping, mutated in place.
|
||||
keys: Program basename to attribution key.
|
||||
"""
|
||||
names = frozenset(keys)
|
||||
for source in (get_pids_by_process_names(names), get_pids_by_cmdline_names(names)):
|
||||
for pid, name in source.items():
|
||||
key = keys.get(name)
|
||||
if key is not None:
|
||||
found.setdefault(pid, key)
|
||||
|
||||
|
||||
def qualifying_pids(rules: PlaytimeRules) -> dict[int, str]:
|
||||
"""Return PIDs whose runtime counts against the daily budget, with owners.
|
||||
|
||||
Steam games are identified by the ``SteamAppId`` environment variable, using
|
||||
the same ``!= 0`` predicate ``enforcer.enforce_allowed_game`` uses to exclude
|
||||
the Steam client tree — browsing the store is not gaming.
|
||||
|
||||
The value is an attribution key (``app:<id>``, ``launcher:<name>`` or
|
||||
``proc:<id>``) so that the budget can record *which* game it billed. The
|
||||
app id was previously destructured and dropped on the first line here, one
|
||||
step before accumulation.
|
||||
|
||||
``counted_processes`` is deliberately *not* gated on ``count_launchers``:
|
||||
that switch covers the sixteen hardcoded launchers, and turning it off must
|
||||
not silently stop billing a game the user explicitly listed.
|
||||
|
||||
Args:
|
||||
rules: Policy for this tick.
|
||||
|
||||
Returns:
|
||||
The set of qualifying PIDs.
|
||||
Mapping of qualifying PID to its attribution key.
|
||||
"""
|
||||
pids = {pid for pid, app_id in get_running_steam_game_pids().items() if app_id != 0}
|
||||
found: dict[int, str] = {
|
||||
pid: f"app:{app_id}"
|
||||
for pid, app_id in get_running_steam_game_pids().items()
|
||||
if app_id != 0
|
||||
}
|
||||
if rules.count_launchers:
|
||||
pids |= set(get_pids_by_process_names(LAUNCHER_PROCESS_NAMES))
|
||||
pids |= set(get_pids_by_cmdline_names(LAUNCHER_PROCESS_NAMES))
|
||||
return pids
|
||||
_merge_named(
|
||||
found, {name: f"launcher:{name}" for name in LAUNCHER_PROCESS_NAMES}
|
||||
)
|
||||
|
||||
counted = key_by_name(load_counted_processes())
|
||||
if counted:
|
||||
_merge_named(found, counted)
|
||||
return found
|
||||
|
||||
|
||||
def attributed_key(qualifying: dict[int, str], focus_pid: int | None) -> str:
|
||||
"""Return the single attribution key this tick should be credited to.
|
||||
|
||||
The focused window decides it, walking ancestry the same way
|
||||
``_engagement_probes.focus_qualifies`` does — a Proton game's window belongs
|
||||
to a child of the process carrying ``SteamAppId``.
|
||||
|
||||
Falling back to "the only game running" keeps attribution working when the
|
||||
focus probe is unavailable (no X, or a toolkit that omits ``_NET_WM_PID``).
|
||||
With two games running and no focus signal there is no honest answer, so it
|
||||
returns ``""``: the tick still bills, and the gap shows up as Unattributed
|
||||
rather than being charged to a guess.
|
||||
|
||||
Args:
|
||||
qualifying: Mapping of qualifying PID to attribution key.
|
||||
focus_pid: PID owning the focused window, or ``None``.
|
||||
|
||||
Returns:
|
||||
The attribution key, or ``""`` if none can be determined.
|
||||
"""
|
||||
if focus_pid is not None:
|
||||
pid = focus_pid
|
||||
for _ in range(_MAX_PROCESS_TREE_DEPTH):
|
||||
key = qualifying.get(pid)
|
||||
if key is not None:
|
||||
return key
|
||||
if pid <= _INIT_PID:
|
||||
break
|
||||
parent = _read_ppid(pid)
|
||||
if parent is None:
|
||||
break
|
||||
pid = parent
|
||||
|
||||
distinct = set(qualifying.values())
|
||||
return distinct.pop() if len(distinct) == 1 else ""
|
||||
|
||||
@ -26,6 +26,10 @@ logger = logging.getLogger(__name__)
|
||||
PLAYTIME_STATE_FILE = CONFIG_DIR / "playtime_state.json"
|
||||
PLAYTIME_DEMO_STATE_FILE = CONFIG_DIR / "playtime_demo_state.json"
|
||||
|
||||
# Unchanged by per-game attribution: `load_state` keeps only keys that are
|
||||
# dataclass fields, so a record written before `per_game` existed loads with
|
||||
# the field's empty default. A bump would buy nothing and would route the
|
||||
# live file through the fail-closed unknown-schema path.
|
||||
_SCHEMA_VERSION = 1
|
||||
|
||||
# The daemon runs as root, and ``mkstemp`` would leave this at 0600 — which
|
||||
@ -72,6 +76,21 @@ class PlaytimeState:
|
||||
"""Epoch seconds of the last accounting tick; ``0.0`` means never ticked."""
|
||||
blocked_at: float = 0.0
|
||||
"""Epoch seconds when the cutoff engaged; ``0.0`` means not blocked."""
|
||||
per_game: dict[str, float] = field(default_factory=dict)
|
||||
"""Attribution key to seconds credited during ``day_key``.
|
||||
|
||||
Only ever a *subset* of ``seconds``: a tick that cannot be attributed to one
|
||||
game still bills, and ``backdate`` refunds can outpace what one key holds.
|
||||
The difference is rendered as "Unattributed" rather than stored, so the
|
||||
invariant is ``sum(per_game.values()) <= seconds`` — never equality.
|
||||
"""
|
||||
last_credited_key: str = ""
|
||||
"""Key credited by the most recent billing tick.
|
||||
|
||||
Exists so ``backdate``'s idle-grace refund can debit the same key it
|
||||
credited; without it every engaged-to-paused edge would drift the parts
|
||||
above the whole.
|
||||
"""
|
||||
warned_seconds: list[int] = field(default_factory=list)
|
||||
"""Warning thresholds already fired during ``day_key``."""
|
||||
|
||||
|
||||
@ -10,7 +10,7 @@ import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer._allowed_games import allowed_games
|
||||
from steam_backlog_enforcer.config import load_snapshot
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer.enforcer import send_notification
|
||||
from steam_backlog_enforcer.game_install import _echo
|
||||
from steam_backlog_enforcer.steam_api import SteamAPIClient
|
||||
|
||||
40
steam_backlog_enforcer/_snapshot.py
Normal file
40
steam_backlog_enforcer/_snapshot.py
Normal file
@ -0,0 +1,40 @@
|
||||
"""Persistence for the cached achievement snapshot.
|
||||
|
||||
Split out of :mod:`steam_backlog_enforcer.config` when that module reached the
|
||||
250-line cap. The dependency runs one way — this module imports the paths from
|
||||
``config``, never the reverse — so callers import these two names from here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from steam_backlog_enforcer import config
|
||||
from steam_backlog_enforcer.config import _atomic_write
|
||||
|
||||
|
||||
def save_snapshot(data: list[dict[str, Any]]) -> None:
|
||||
"""Save an achievement snapshot to disk.
|
||||
|
||||
Args:
|
||||
data: The snapshot rows to persist.
|
||||
"""
|
||||
_atomic_write(
|
||||
config.SNAPSHOT_FILE,
|
||||
json.dumps(data, indent=2) + "\n",
|
||||
)
|
||||
|
||||
|
||||
def load_snapshot() -> list[dict[str, Any]] | None:
|
||||
"""Load the cached achievement snapshot.
|
||||
|
||||
Returns:
|
||||
The snapshot rows, or ``None`` when no snapshot has been written.
|
||||
"""
|
||||
if config.SNAPSHOT_FILE.exists():
|
||||
result: list[dict[str, Any]] = json.loads(
|
||||
config.SNAPSHOT_FILE.read_text(encoding="utf-8")
|
||||
)
|
||||
return result
|
||||
return None
|
||||
@ -8,6 +8,7 @@ from typing import TYPE_CHECKING
|
||||
from steam_backlog_enforcer._hltb_types import (
|
||||
_read_raw_cache,
|
||||
)
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer._stats_display import (
|
||||
_format_completion_date,
|
||||
_print_pace_scenario,
|
||||
@ -26,7 +27,6 @@ from steam_backlog_enforcer._web_dataset import (
|
||||
compute_pace_vs_hltb,
|
||||
count_complete_since_start,
|
||||
)
|
||||
from steam_backlog_enforcer.config import load_snapshot
|
||||
from steam_backlog_enforcer.game_install import _echo
|
||||
from steam_backlog_enforcer.steam_api import (
|
||||
GameInfo,
|
||||
|
||||
@ -24,6 +24,7 @@ import logging
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
from steam_backlog_enforcer._counted_procs import kill_target_names
|
||||
from steam_backlog_enforcer._total_block_hosts import (
|
||||
apply_total_block_hosts,
|
||||
remove_total_block_hosts,
|
||||
@ -41,6 +42,7 @@ from steam_backlog_enforcer._total_block_purge import (
|
||||
uninstall_steam_package,
|
||||
)
|
||||
from steam_backlog_enforcer.config import CONFIG_DIR
|
||||
from steam_backlog_enforcer.enforcer import kill_processes_by_name
|
||||
from steam_backlog_enforcer.store_blocker import flush_dns_cache
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -177,6 +179,9 @@ def start_total_block(days: int) -> bool:
|
||||
return False
|
||||
|
||||
killed = kill_steam_and_launchers(LAUNCHER_PROCESS_NAMES)
|
||||
# Kill only: kill_steam_and_launchers() uninstalls what it is given, and a
|
||||
# counted process is a game the user chose to bill, not a launcher to purge.
|
||||
killed += kill_processes_by_name(kill_target_names())
|
||||
if killed:
|
||||
logger.info("Total block: killed %d process(es): %s", len(killed), killed)
|
||||
|
||||
@ -204,6 +209,7 @@ def enforce_total_block_tick() -> None:
|
||||
is True.
|
||||
"""
|
||||
kill_steam_and_launchers(LAUNCHER_PROCESS_NAMES)
|
||||
kill_processes_by_name(kill_target_names())
|
||||
|
||||
if is_steam_installed():
|
||||
logger.warning("Steam reappeared during total block - removing again")
|
||||
|
||||
@ -38,7 +38,7 @@ __all__ = [
|
||||
|
||||
from dataclasses import asdict
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from steam_backlog_enforcer._actions import allowed_app_ids
|
||||
from steam_backlog_enforcer._hltb_types import _read_raw_cache
|
||||
@ -47,6 +47,7 @@ from steam_backlog_enforcer._scanning_confidence import (
|
||||
_MIN_CONFIDENCE_SUM,
|
||||
_MIN_COUNT_COMP,
|
||||
)
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer._web_games import (
|
||||
_build_games,
|
||||
_default_qualifying,
|
||||
@ -66,12 +67,14 @@ from steam_backlog_enforcer._web_models import (
|
||||
WebStateInfo,
|
||||
)
|
||||
from steam_backlog_enforcer._web_pace import compute_pace_vs_hltb
|
||||
from steam_backlog_enforcer.config import State, load_snapshot
|
||||
from steam_backlog_enforcer.protondb import (
|
||||
MIN_PLAYABLE_TIER,
|
||||
)
|
||||
from steam_backlog_enforcer.steam_api import GameInfo
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from steam_backlog_enforcer.config import State
|
||||
|
||||
|
||||
def count_complete_since_start(games: list[GameInfo], started_at: str) -> int:
|
||||
"""Count complete games whose last achievement was unlocked on/after started_at.
|
||||
|
||||
@ -95,6 +95,19 @@ class Config:
|
||||
True is the safer default: under-counting is the failure mode that quietly
|
||||
defeats the budget entirely.
|
||||
"""
|
||||
counted_processes: list[dict[str, Any]] | None = None
|
||||
"""Non-Steam games that count against the budget, beyond the launcher set.
|
||||
|
||||
Each entry is ``{"id": ..., "label": ..., "names": [...]}``. ``None`` means
|
||||
"use the built-in defaults" (osu!lazer); an explicit ``[]`` means "none",
|
||||
which is why this is not simply an empty list. Parsed and validated by
|
||||
:mod:`steam_backlog_enforcer._counted_procs` — the field exists here only so
|
||||
that ``save()`` round-trips a hand-edited list instead of dropping it.
|
||||
|
||||
Deliberately separate from ``LAUNCHER_PROCESS_NAMES``: that frozenset is
|
||||
also the total-block kill list, so adding a game there has side effects
|
||||
this list does not.
|
||||
"""
|
||||
playtime_enforcement: bool = True
|
||||
"""Master switch for the daily gaming budget.
|
||||
|
||||
@ -230,21 +243,3 @@ class State:
|
||||
self.manual_pick_app_id = None
|
||||
self.manual_pick_game_name = ""
|
||||
self.manual_pick_started_at = ""
|
||||
|
||||
|
||||
def save_snapshot(data: list[dict[str, Any]]) -> None:
|
||||
"""Save an achievement snapshot to disk."""
|
||||
_atomic_write(
|
||||
SNAPSHOT_FILE,
|
||||
json.dumps(data, indent=2) + "\n",
|
||||
)
|
||||
|
||||
|
||||
def load_snapshot() -> list[dict[str, Any]] | None:
|
||||
"""Load the cached achievement snapshot, or None if absent."""
|
||||
if SNAPSHOT_FILE.exists():
|
||||
result: list[dict[str, Any]] = json.loads(
|
||||
SNAPSHOT_FILE.read_text(encoding="utf-8")
|
||||
)
|
||||
return result
|
||||
return None
|
||||
|
||||
@ -21,7 +21,7 @@ from steam_backlog_enforcer._pick_completion import (
|
||||
report_completion,
|
||||
warn_stale_assignment,
|
||||
)
|
||||
from steam_backlog_enforcer.config import load_snapshot
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer.game_install import (
|
||||
_echo,
|
||||
install_game,
|
||||
|
||||
@ -9,8 +9,8 @@ from steam_backlog_enforcer._actions import (
|
||||
manual_pick_age_days,
|
||||
)
|
||||
from steam_backlog_enforcer._pick_completion import report_completion
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot
|
||||
from steam_backlog_enforcer._total_block import get_total_block_status
|
||||
from steam_backlog_enforcer.config import load_snapshot
|
||||
from steam_backlog_enforcer.game_install import (
|
||||
_echo,
|
||||
get_installed_games,
|
||||
|
||||
@ -32,12 +32,7 @@ from steam_backlog_enforcer._scanning_tampering import (
|
||||
_check_game_tampering,
|
||||
detect_tampering,
|
||||
)
|
||||
from steam_backlog_enforcer.config import (
|
||||
Config,
|
||||
State,
|
||||
load_snapshot,
|
||||
save_snapshot,
|
||||
)
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot, save_snapshot
|
||||
from steam_backlog_enforcer.enforcer import (
|
||||
send_notification,
|
||||
)
|
||||
@ -52,6 +47,8 @@ from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
|
||||
# These helpers moved to the _scanning_* leaf modules, but callers and tests
|
||||
# have always imported them from here. mypy's --no-implicit-reexport needs
|
||||
# them listed explicitly.
|
||||
|
||||
79
steam_backlog_enforcer/tests/test_attribution_labels.py
Normal file
79
steam_backlog_enforcer/tests/test_attribution_labels.py
Normal file
@ -0,0 +1,79 @@
|
||||
"""Tests for _attribution_labels — 100% branch coverage."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer import _attribution_labels as labels_mod
|
||||
from steam_backlog_enforcer import _steam_state
|
||||
from steam_backlog_enforcer._attribution_labels import label_for, labels_for
|
||||
from steam_backlog_enforcer._counted_procs import CountedProcess
|
||||
|
||||
|
||||
def _manifest(app_id: int, name: str) -> None:
|
||||
_steam_state.STEAMAPPS_PATH.mkdir(parents=True, exist_ok=True)
|
||||
(_steam_state.STEAMAPPS_PATH / f"appmanifest_{app_id}.acf").write_text(
|
||||
f'"AppState"\n{{\n\t"appid"\t\t"{app_id}"\n\t"name"\t\t"{name}"\n}}\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
|
||||
class TestLabelFor:
|
||||
"""Resolving each key shape."""
|
||||
|
||||
def setup_method(self) -> None:
|
||||
labels_mod._app_names.clear()
|
||||
|
||||
def test_reads_a_steam_name_from_the_appmanifest(self) -> None:
|
||||
# A running game is by definition installed, so its manifest exists —
|
||||
# which is why this never has to load the 11 MB snapshot.
|
||||
_manifest(475150, "Titan Quest Anniversary Edition")
|
||||
assert label_for("app:475150") == "Titan Quest Anniversary Edition"
|
||||
|
||||
def test_memoises_the_manifest_read(self) -> None:
|
||||
_manifest(440, "Team Fortress 2")
|
||||
assert label_for("app:440") == "Team Fortress 2"
|
||||
with patch.object(_steam_state, "STEAMAPPS_PATH", None):
|
||||
# A second lookup must not touch the filesystem at all.
|
||||
assert label_for("app:440") == "Team Fortress 2"
|
||||
|
||||
def test_a_missing_manifest_falls_back_to_the_key(self) -> None:
|
||||
assert label_for("app:999999") == "app:999999"
|
||||
|
||||
def test_a_manifest_without_a_name_falls_back_to_the_key(self) -> None:
|
||||
_steam_state.STEAMAPPS_PATH.mkdir(parents=True, exist_ok=True)
|
||||
(_steam_state.STEAMAPPS_PATH / "appmanifest_5.acf").write_text(
|
||||
'"AppState"\n{\n\t"appid"\t\t"5"\n}\n', encoding="utf-8"
|
||||
)
|
||||
assert label_for("app:5") == "app:5"
|
||||
|
||||
def test_a_non_numeric_app_key_falls_back(self) -> None:
|
||||
assert label_for("app:nope") == "app:nope"
|
||||
|
||||
def test_a_counted_process_uses_its_configured_label(self) -> None:
|
||||
with patch.object(
|
||||
labels_mod,
|
||||
"load_counted_processes",
|
||||
return_value=(
|
||||
CountedProcess(id="osu-lazer", label="osu!lazer", names=frozenset()),
|
||||
),
|
||||
):
|
||||
assert label_for("proc:osu-lazer") == "osu!lazer"
|
||||
|
||||
def test_an_unknown_counted_process_falls_back(self) -> None:
|
||||
with patch.object(labels_mod, "load_counted_processes", return_value=()):
|
||||
assert label_for("proc:gone") == "proc:gone"
|
||||
|
||||
def test_a_launcher_uses_its_process_name(self) -> None:
|
||||
assert label_for("launcher:lutris") == "lutris"
|
||||
|
||||
def test_an_unrecognised_key_is_returned_as_is(self) -> None:
|
||||
assert label_for("weird") == "weird"
|
||||
|
||||
|
||||
class TestLabelsFor:
|
||||
def test_resolves_every_key(self) -> None:
|
||||
assert labels_for(["launcher:heroic", "launcher:itch"]) == {
|
||||
"launcher:heroic": "heroic",
|
||||
"launcher:itch": "itch",
|
||||
}
|
||||
122
steam_backlog_enforcer/tests/test_budget_games.py
Normal file
122
steam_backlog_enforcer/tests/test_budget_games.py
Normal file
@ -0,0 +1,122 @@
|
||||
"""Tests for _budget_games — 100% branch coverage.
|
||||
|
||||
The invariant under test is that the parts never claim more than the whole:
|
||||
`per_game` is a subset of `seconds`, and the difference is *rendered* as
|
||||
Unattributed rather than stored or hidden.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from steam_backlog_enforcer import _budget_games as games_mod
|
||||
from steam_backlog_enforcer._budget_games import history_view, today_games
|
||||
from steam_backlog_enforcer._playtime_history import HistoryDay
|
||||
from steam_backlog_enforcer._playtime_state import PlaytimeState
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Iterator
|
||||
|
||||
_LABELS = {"app:1": "One", "app:2": "Two", "proc:osu-lazer": "osu!lazer"}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _labels() -> Iterator[None]:
|
||||
"""Serve the persisted label map without touching the history file."""
|
||||
with patch.object(games_mod, "load_labels", return_value=_LABELS):
|
||||
yield
|
||||
|
||||
|
||||
class TestTodayGames:
|
||||
"""Today's list is uncapped: it is text, not six colours."""
|
||||
|
||||
def test_orders_largest_first_with_shares(self) -> None:
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-29",
|
||||
seconds=300.0,
|
||||
per_game={"app:1": 100.0, "app:2": 200.0},
|
||||
)
|
||||
assert today_games(state) == [
|
||||
{"key": "app:2", "label": "Two", "seconds": 200.0, "fraction": 0.6667},
|
||||
{"key": "app:1", "label": "One", "seconds": 100.0, "fraction": 0.3333},
|
||||
]
|
||||
|
||||
def test_surfaces_the_unattributed_remainder_last(self) -> None:
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-29", seconds=300.0, per_game={"app:1": 200.0}
|
||||
)
|
||||
assert today_games(state)[-1] == {
|
||||
"key": "unattributed",
|
||||
"label": "Unattributed",
|
||||
"seconds": 100.0,
|
||||
"fraction": 0.3333,
|
||||
}
|
||||
|
||||
def test_no_remainder_when_fully_attributed(self) -> None:
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-29", seconds=100.0, per_game={"app:1": 100.0}
|
||||
)
|
||||
assert [g["key"] for g in today_games(state)] == ["app:1"]
|
||||
|
||||
def test_a_day_that_billed_nothing_lists_nothing(self) -> None:
|
||||
assert today_games(PlaytimeState(day_key="2026-08-29")) == []
|
||||
|
||||
def test_drift_below_zero_cannot_produce_a_negative_slice(self) -> None:
|
||||
"""Proportional refunds drift by fractions of a second."""
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-29", seconds=100.0, per_game={"app:1": 100.5}
|
||||
)
|
||||
assert [g["key"] for g in today_games(state)] == ["app:1"]
|
||||
|
||||
def test_falls_back_to_a_live_lookup_for_an_unstored_key(self) -> None:
|
||||
"""Today's keys reach the history file only after the first flush."""
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-29", seconds=10.0, per_game={"launcher:lutris": 10.0}
|
||||
)
|
||||
assert today_games(state)[0]["label"] == "lutris"
|
||||
|
||||
|
||||
class TestHistoryView:
|
||||
"""The window is capped at the six hues the shared ramp actually has."""
|
||||
|
||||
def test_orders_segments_by_the_shared_legend(self) -> None:
|
||||
days = [
|
||||
HistoryDay("2026-08-27", 300.0, {"app:1": 100.0, "app:2": 200.0}),
|
||||
HistoryDay("2026-08-28", 50.0, {"app:1": 50.0}),
|
||||
]
|
||||
# Ranked by window total (app:2 200s, app:1 150s), not per-day, so a
|
||||
# game keeps one colour and one band position across every bar.
|
||||
plotted, legend = history_view(days)
|
||||
assert [e["key"] for e in legend] == ["app:2", "app:1"]
|
||||
assert plotted[0]["segments"] == [
|
||||
{"key": "app:2", "seconds": 200.0},
|
||||
{"key": "app:1", "seconds": 100.0},
|
||||
]
|
||||
assert plotted[1]["segments"] == [{"key": "app:1", "seconds": 50.0}]
|
||||
|
||||
def test_folds_the_seventh_game_into_other(self) -> None:
|
||||
games = {f"app:{i}": float(100 - i) for i in range(8)}
|
||||
plotted, legend = history_view(
|
||||
[HistoryDay("2026-08-28", sum(games.values()), games)]
|
||||
)
|
||||
assert [e["key"] for e in legend][-1] == "other"
|
||||
assert len([e for e in legend if e["key"] != "other"]) == 6
|
||||
# Folding must conserve time, not drop it.
|
||||
assert sum(s["seconds"] for s in plotted[0]["segments"]) == sum(games.values())
|
||||
|
||||
def test_a_legacy_day_is_entirely_unattributed(self) -> None:
|
||||
"""Schema-1 days stored a bare total and never had a breakdown."""
|
||||
plotted, legend = history_view([HistoryDay("2026-08-24", 1200.0, {})])
|
||||
assert plotted[0]["segments"] == [{"key": "unattributed", "seconds": 1200.0}]
|
||||
assert legend == [{"key": "unattributed", "label": "Unattributed"}]
|
||||
|
||||
def test_a_day_with_no_time_has_no_segments(self) -> None:
|
||||
plotted, legend = history_view([HistoryDay("2026-08-24", 0.0, {})])
|
||||
assert plotted[0]["segments"] == []
|
||||
assert legend == []
|
||||
|
||||
def test_empty_window(self) -> None:
|
||||
assert history_view([]) == ([], [])
|
||||
@ -178,7 +178,16 @@ class TestBuildBudgetSnapshot:
|
||||
return_value=[HistoryDay(day="2026-08-27", seconds=1.5)],
|
||||
):
|
||||
snapshot = self._snapshot()
|
||||
assert snapshot["history"] == [{"day": "2026-08-27", "seconds": 1.5}]
|
||||
# A day with a total but no breakdown is entirely unattributed: the
|
||||
# residual is computed here, never stored.
|
||||
assert snapshot["history"] == [
|
||||
{
|
||||
"day": "2026-08-27",
|
||||
"seconds": 1.5,
|
||||
"segments": [{"key": "unattributed", "seconds": 1.5}],
|
||||
}
|
||||
]
|
||||
assert snapshot["legend"] == [{"key": "unattributed", "label": "Unattributed"}]
|
||||
|
||||
def test_demo_runs_serve_no_history(self) -> None:
|
||||
"""Production days must not be plotted against the 60-second budget."""
|
||||
|
||||
@ -7,12 +7,8 @@ import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer.config import (
|
||||
Config,
|
||||
State,
|
||||
load_snapshot,
|
||||
save_snapshot,
|
||||
)
|
||||
from steam_backlog_enforcer._snapshot import load_snapshot, save_snapshot
|
||||
from steam_backlog_enforcer.config import Config, State
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
146
steam_backlog_enforcer/tests/test_counted_procs.py
Normal file
146
steam_backlog_enforcer/tests/test_counted_procs.py
Normal file
@ -0,0 +1,146 @@
|
||||
"""Tests for _counted_procs — 100% branch coverage.
|
||||
|
||||
The validation here is load-bearing rather than cosmetic: these names are fed
|
||||
to the budget cutoff and the total-block killer, and `_playtime_kill` signals a
|
||||
matched PID *and its descendants*. A name like `sh` would take the whole
|
||||
session down with it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from steam_backlog_enforcer import config as config_mod
|
||||
from steam_backlog_enforcer._counted_procs import (
|
||||
DEFAULT_COUNTED_PROCESSES,
|
||||
CountedProcess,
|
||||
counted_process_names,
|
||||
key_by_name,
|
||||
kill_target_names,
|
||||
labels_by_key,
|
||||
load_counted_processes,
|
||||
parse_counted_processes,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _write(payload: object) -> None:
|
||||
config_mod.CONFIG_FILE.write_text(json.dumps(payload), encoding="utf-8")
|
||||
|
||||
|
||||
class TestParse:
|
||||
"""Turning raw JSON into validated entries."""
|
||||
|
||||
def test_reads_a_well_formed_entry(self) -> None:
|
||||
entries = parse_counted_processes(
|
||||
[{"id": "osu-lazer", "label": "osu!lazer", "names": ["osu!", "osu-lazer"]}]
|
||||
)
|
||||
assert entries == (
|
||||
CountedProcess(
|
||||
id="osu-lazer",
|
||||
label="osu!lazer",
|
||||
names=frozenset({"osu!", "osu-lazer"}),
|
||||
),
|
||||
)
|
||||
|
||||
def test_label_defaults_to_the_id(self) -> None:
|
||||
entries = parse_counted_processes([{"id": "quake", "names": ["quake"]}])
|
||||
assert entries[0].label == "quake"
|
||||
|
||||
def test_rejects_a_non_list(self) -> None:
|
||||
assert parse_counted_processes({"id": "x"}) == ()
|
||||
|
||||
def test_drops_a_non_object_entry(self) -> None:
|
||||
assert parse_counted_processes(["nope"]) == ()
|
||||
|
||||
def test_drops_an_entry_with_no_id(self) -> None:
|
||||
assert parse_counted_processes([{"names": ["x"]}]) == ()
|
||||
|
||||
def test_drops_an_entry_with_no_usable_names(self) -> None:
|
||||
assert parse_counted_processes([{"id": "x", "names": []}]) == ()
|
||||
|
||||
def test_drops_an_entry_whose_names_are_not_a_list(self) -> None:
|
||||
assert parse_counted_processes([{"id": "x", "names": "osu!"}]) == ()
|
||||
|
||||
def test_drops_a_duplicate_id(self) -> None:
|
||||
entries = parse_counted_processes(
|
||||
[
|
||||
{"id": "x", "names": ["a"]},
|
||||
{"id": "x", "names": ["b"]},
|
||||
]
|
||||
)
|
||||
assert len(entries) == 1
|
||||
assert entries[0].names == frozenset({"a"})
|
||||
|
||||
def test_ignores_non_string_and_blank_names(self) -> None:
|
||||
entries = parse_counted_processes([{"id": "x", "names": ["a", 7, " ", None]}])
|
||||
assert entries[0].names == frozenset({"a"})
|
||||
|
||||
def test_refuses_a_shell_or_interpreter_name(self) -> None:
|
||||
"""`sh` would SIGTERM every shell on the system, and its children."""
|
||||
entries = parse_counted_processes(
|
||||
[{"id": "osu", "names": ["sh", "dotnet", "osu!"]}]
|
||||
)
|
||||
assert entries[0].names == frozenset({"osu!"})
|
||||
|
||||
def test_an_entry_of_only_forbidden_names_is_dropped(self) -> None:
|
||||
assert parse_counted_processes([{"id": "osu", "names": ["sh"]}]) == ()
|
||||
|
||||
|
||||
class TestLoad:
|
||||
"""Reading straight from config.json, with no Config to consult."""
|
||||
|
||||
def test_missing_file_falls_back_to_defaults(self) -> None:
|
||||
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
|
||||
|
||||
def test_unreadable_file_falls_back_to_defaults(self) -> None:
|
||||
config_mod.CONFIG_FILE.write_text("{not json", encoding="utf-8")
|
||||
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
|
||||
|
||||
def test_non_object_payload_falls_back_to_defaults(self) -> None:
|
||||
_write([1, 2, 3])
|
||||
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
|
||||
|
||||
def test_absent_key_falls_back_to_defaults(self) -> None:
|
||||
_write({"steam_id": "1"})
|
||||
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
|
||||
|
||||
def test_an_explicit_empty_list_means_none(self) -> None:
|
||||
"""Distinct from absent: the user can switch the feature off."""
|
||||
_write({"counted_processes": []})
|
||||
assert load_counted_processes() == ()
|
||||
|
||||
def test_reads_a_configured_list(self) -> None:
|
||||
_write({"counted_processes": [{"id": "q", "names": ["quake"]}]})
|
||||
assert load_counted_processes()[0].id == "q"
|
||||
|
||||
def test_the_default_covers_every_osu_layer(self) -> None:
|
||||
"""The sh wrapper, the AppImage and the game each exec a new comm."""
|
||||
assert counted_process_names(DEFAULT_COUNTED_PROCESSES) == frozenset(
|
||||
{"osu-lazer", "osu.AppImage", "osu!"}
|
||||
)
|
||||
|
||||
|
||||
class TestDerivedViews:
|
||||
"""The three projections the billing and kill paths consume."""
|
||||
|
||||
_ENTRIES = (
|
||||
CountedProcess(id="osu-lazer", label="osu!lazer", names=frozenset({"osu!"})),
|
||||
)
|
||||
|
||||
def test_names_of_no_entries_is_empty(self) -> None:
|
||||
assert counted_process_names(()) == frozenset()
|
||||
|
||||
def test_key_by_name(self) -> None:
|
||||
assert key_by_name(self._ENTRIES) == {"osu!": "proc:osu-lazer"}
|
||||
|
||||
def test_labels_by_key(self) -> None:
|
||||
assert labels_by_key(self._ENTRIES) == {"proc:osu-lazer": "osu!lazer"}
|
||||
|
||||
def test_kill_targets_come_from_the_config(self, tmp_path: Path) -> None:
|
||||
assert tmp_path.exists()
|
||||
_write({"counted_processes": [{"id": "q", "names": ["quake"]}]})
|
||||
assert kill_target_names() == frozenset({"quake"})
|
||||
212
steam_backlog_enforcer/tests/test_playtime_attribution.py
Normal file
212
steam_backlog_enforcer/tests/test_playtime_attribution.py
Normal file
@ -0,0 +1,212 @@
|
||||
"""Tests for who a billed tick belongs to — 100% branch coverage.
|
||||
|
||||
Split from ``test_playtime_procs`` at the 250-line cap. This file owns the
|
||||
attribution dimension: which PIDs qualify, and which single game a tick is
|
||||
credited to. ``test_playtime_procs`` keeps the /proc plumbing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._counted_procs import CountedProcess
|
||||
from steam_backlog_enforcer._playtime_procs import attributed_key, qualifying_pids
|
||||
from steam_backlog_enforcer._playtime_state import rules_for
|
||||
from steam_backlog_enforcer.config import Config
|
||||
|
||||
|
||||
class TestQualifyingPids:
|
||||
"""`qualifying_pids` returns pid -> attribution key, not a bare pid set."""
|
||||
|
||||
def test_steam_games_only_when_launchers_are_off(self) -> None:
|
||||
rules = rules_for(Config(count_launcher_processes=False), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={5: 440, 6: 0},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
|
||||
return_value=(),
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names"
|
||||
) as mock_comm,
|
||||
):
|
||||
assert qualifying_pids(rules) == {5: "app:440"}
|
||||
mock_comm.assert_not_called()
|
||||
|
||||
def test_steam_client_appid_zero_does_not_count(self) -> None:
|
||||
"""Browsing the store is not gaming."""
|
||||
rules = rules_for(Config(count_launcher_processes=False), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={9: 0},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
|
||||
return_value=(),
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {}
|
||||
|
||||
def test_unions_launchers_from_both_scanners(self) -> None:
|
||||
rules = rules_for(Config(count_launcher_processes=True), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={5: 440},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
|
||||
return_value=(),
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={7: "heroic"},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={8: "lutris"},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {
|
||||
5: "app:440",
|
||||
7: "launcher:heroic",
|
||||
8: "launcher:lutris",
|
||||
}
|
||||
|
||||
def test_counted_processes_bill_even_with_launchers_off(self) -> None:
|
||||
"""The user's own list is not gated on the launcher switch.
|
||||
|
||||
Turning launcher counting off because a launcher idles in the tray must
|
||||
not silently stop billing a game the user explicitly listed.
|
||||
"""
|
||||
rules = rules_for(Config(count_launcher_processes=False), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
|
||||
return_value=(
|
||||
CountedProcess(
|
||||
id="osu-lazer", label="osu!lazer", names=frozenset({"osu!"})
|
||||
),
|
||||
),
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={11: "osu!"},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {11: "proc:osu-lazer"}
|
||||
|
||||
def test_steam_attribution_wins_over_a_name_match(self) -> None:
|
||||
"""A Steam game keeps its app key even if its comm matches a launcher."""
|
||||
rules = rules_for(Config(count_launcher_processes=True), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={7: 440},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
|
||||
return_value=(),
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={7: "heroic"},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {7: "app:440"}
|
||||
|
||||
def test_a_name_outside_the_map_is_skipped(self) -> None:
|
||||
"""Unreachable in production; a KeyError in the enforce loop is not."""
|
||||
rules = rules_for(Config(count_launcher_processes=True), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
|
||||
return_value=(),
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={12: "not-a-launcher"},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {}
|
||||
|
||||
def test_no_processes_at_all(self) -> None:
|
||||
rules = rules_for(Config(), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {}
|
||||
|
||||
|
||||
class TestAttributedKey:
|
||||
"""Deciding which single game a tick belongs to."""
|
||||
|
||||
_PKG = "steam_backlog_enforcer._playtime_procs"
|
||||
|
||||
def test_focused_pid_wins(self) -> None:
|
||||
assert attributed_key({7: "app:440", 8: "proc:osu"}, 8) == "proc:osu"
|
||||
|
||||
def test_walks_up_to_the_owning_process(self) -> None:
|
||||
"""A Proton game's window belongs to a child of the SteamAppId process."""
|
||||
with patch(f"{self._PKG}._read_ppid", side_effect=[900, 7]):
|
||||
assert attributed_key({7: "app:440"}, 999) == "app:440"
|
||||
|
||||
def test_gives_up_when_the_chain_ends(self) -> None:
|
||||
with patch(f"{self._PKG}._read_ppid", return_value=None):
|
||||
assert attributed_key({7: "app:440", 8: "app:1"}, 999) == ""
|
||||
|
||||
def test_stops_at_init(self) -> None:
|
||||
with patch(f"{self._PKG}._read_ppid", return_value=1):
|
||||
assert attributed_key({7: "app:440", 8: "app:1"}, 999) == ""
|
||||
|
||||
def test_stops_after_the_depth_limit(self) -> None:
|
||||
"""A ppid cycle must not spin the enforce loop."""
|
||||
with patch(f"{self._PKG}._read_ppid", return_value=500):
|
||||
assert attributed_key({7: "app:440", 8: "app:1"}, 500) == ""
|
||||
|
||||
def test_falls_back_to_the_only_game_running(self) -> None:
|
||||
"""Covers no X, or a toolkit that omits _NET_WM_PID."""
|
||||
assert attributed_key({7: "app:440", 8: "app:440"}, None) == "app:440"
|
||||
|
||||
def test_two_games_and_no_focus_is_unattributable(self) -> None:
|
||||
# There is no honest answer, so the tick still bills but the gap shows
|
||||
# up as Unattributed rather than being charged to a guess.
|
||||
assert attributed_key({7: "app:440", 8: "app:1"}, None) == ""
|
||||
|
||||
def test_nothing_qualifying(self) -> None:
|
||||
assert attributed_key({}, None) == ""
|
||||
@ -128,6 +128,47 @@ class TestBackdate:
|
||||
state = tracker.backdate(state, _idle_paused(), rules=RULES)
|
||||
assert state.seconds == 700.0
|
||||
|
||||
def test_debits_the_refund_from_the_credited_game(self) -> None:
|
||||
"""Otherwise the parts drift above the whole on every walk-away.
|
||||
|
||||
This edge fires many times a day, so leaving `per_game` untouched would
|
||||
make stacked segments overflow their bar within minutes.
|
||||
"""
|
||||
tracker = _tracker()
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-28",
|
||||
seconds=1000.0,
|
||||
per_game={"app:440": 1000.0},
|
||||
last_credited_key="app:440",
|
||||
)
|
||||
state = tracker.backdate(state, _engaged(), rules=RULES)
|
||||
state = tracker.backdate(state, _idle_paused(), rules=RULES)
|
||||
assert state.seconds == 700.0
|
||||
assert state.per_game == {"app:440": 700.0}
|
||||
assert sum(state.per_game.values()) <= state.seconds
|
||||
|
||||
def test_a_refund_larger_than_the_key_holds_floors_at_zero(self) -> None:
|
||||
tracker = _tracker()
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-28",
|
||||
seconds=1000.0,
|
||||
per_game={"app:440": 100.0},
|
||||
last_credited_key="app:440",
|
||||
)
|
||||
state = tracker.backdate(state, _engaged(), rules=RULES)
|
||||
state = tracker.backdate(state, _idle_paused(), rules=RULES)
|
||||
assert state.per_game == {"app:440": 0.0}
|
||||
|
||||
def test_an_unattributed_refund_leaves_the_breakdown_alone(self) -> None:
|
||||
"""Nothing was credited, so there is nothing to take back."""
|
||||
tracker = _tracker()
|
||||
state = PlaytimeState(
|
||||
day_key="2026-08-28", seconds=1000.0, per_game={"app:440": 100.0}
|
||||
)
|
||||
state = tracker.backdate(state, _engaged(), rules=RULES)
|
||||
state = tracker.backdate(state, _idle_paused(), rules=RULES)
|
||||
assert state.per_game == {"app:440": 100.0}
|
||||
|
||||
def test_never_refunds_below_zero(self) -> None:
|
||||
tracker = _tracker()
|
||||
state = tracker.backdate(self._state(seconds=10.0), _engaged(), rules=RULES)
|
||||
|
||||
@ -16,27 +16,32 @@ from steam_backlog_enforcer._playtime_state import PlaytimeState
|
||||
_PKG = "steam_backlog_enforcer._playtime_history"
|
||||
|
||||
|
||||
def _raw() -> dict:
|
||||
return json.loads(history_mod.HISTORY_FILE.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def _stored() -> dict[str, float]:
|
||||
return json.loads(history_mod.HISTORY_FILE.read_text(encoding="utf-8"))["days"]
|
||||
"""Day totals, with the per-game breakdown unwrapped away."""
|
||||
return {day: entry["seconds"] for day, entry in _raw()["days"].items()}
|
||||
|
||||
|
||||
class TestRecordDay:
|
||||
"""Tests for record_day."""
|
||||
|
||||
def test_creates_and_upserts(self) -> None:
|
||||
record_day("2026-08-27", 100.0)
|
||||
record_day("2026-08-28", 200.0)
|
||||
record_day("2026-08-28", 250.0)
|
||||
record_day("2026-08-27", 100.0, {})
|
||||
record_day("2026-08-28", 200.0, {})
|
||||
record_day("2026-08-28", 250.0, {})
|
||||
assert _stored() == {"2026-08-27": 100.0, "2026-08-28": 250.0}
|
||||
|
||||
def test_written_world_readable(self) -> None:
|
||||
record_day("2026-08-28", 1.0)
|
||||
record_day("2026-08-28", 1.0, {})
|
||||
assert history_mod.HISTORY_FILE.stat().st_mode & 0o777 == 0o644
|
||||
|
||||
def test_trims_to_the_retention_window(self) -> None:
|
||||
with patch(f"{_PKG}._MAX_DAYS", 2):
|
||||
for day in ("2026-08-26", "2026-08-27", "2026-08-28"):
|
||||
record_day(day, 1.0)
|
||||
record_day(day, 1.0, {})
|
||||
assert sorted(_stored()) == ["2026-08-27", "2026-08-28"]
|
||||
|
||||
|
||||
@ -52,13 +57,13 @@ class TestLoadHistory:
|
||||
("2026-08-27", 2.0),
|
||||
("2026-08-28", 3.0),
|
||||
):
|
||||
record_day(day, seconds)
|
||||
record_day(day, seconds, {})
|
||||
days = load_history(2)
|
||||
assert [d.day for d in days] == ["2026-08-27", "2026-08-28"]
|
||||
assert days[0].seconds == 2.0
|
||||
|
||||
def test_zero_limit_returns_nothing(self) -> None:
|
||||
record_day("2026-08-28", 1.0)
|
||||
record_day("2026-08-28", 1.0, {})
|
||||
assert load_history(0) == []
|
||||
|
||||
def test_unparseable_file_is_ignored(self) -> None:
|
||||
@ -66,7 +71,7 @@ class TestLoadHistory:
|
||||
assert load_history() == []
|
||||
|
||||
def test_unreadable_file_is_ignored(self) -> None:
|
||||
record_day("2026-08-28", 1.0)
|
||||
record_day("2026-08-28", 1.0, {})
|
||||
with patch("pathlib.Path.read_text", side_effect=PermissionError):
|
||||
assert load_history() == []
|
||||
|
||||
@ -153,3 +158,74 @@ class TestHistoryWriter:
|
||||
# Not remembered as written, so the next attempt tries again.
|
||||
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=False)
|
||||
assert _stored() == {"2026-08-28": 10.0}
|
||||
|
||||
|
||||
class TestPerGameBreakdown:
|
||||
"""The per-game dimension, and the legacy shape it had to grow out of."""
|
||||
|
||||
def test_round_trips_games_and_labels(self) -> None:
|
||||
record_day("2026-08-28", 300.0, {"proc:osu-lazer": 200.0})
|
||||
stored = _raw()
|
||||
assert stored["days"]["2026-08-28"]["games"] == {"proc:osu-lazer": 200.0}
|
||||
assert stored["labels"]["proc:osu-lazer"] == "osu!lazer"
|
||||
|
||||
def test_a_later_flush_preserves_other_days(self) -> None:
|
||||
"""Regression: the writer read through a lossy view and rewrote it.
|
||||
|
||||
A whole-file read-modify-write built from a reader that dropped the
|
||||
`games` sub-dict destroyed every other day's breakdown roughly once a
|
||||
minute, so a bar kept its total and silently lost its bands.
|
||||
"""
|
||||
record_day("2026-08-27", 300.0, {"proc:osu-lazer": 300.0})
|
||||
record_day("2026-08-28", 100.0, {"app:440": 100.0})
|
||||
assert _raw()["days"]["2026-08-27"]["games"] == {"proc:osu-lazer": 300.0}
|
||||
|
||||
def test_reads_a_schema_1_bare_float(self) -> None:
|
||||
"""Days recorded before attribution existed keep their total."""
|
||||
history_mod.HISTORY_FILE.write_text(
|
||||
json.dumps({"schema_version": 1, "days": {"2026-08-20": 1234.5}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
days = load_history()
|
||||
assert [(d.day, d.seconds, d.games) for d in days] == [
|
||||
("2026-08-20", 1234.5, {})
|
||||
]
|
||||
|
||||
def test_drops_malformed_games_and_labels(self) -> None:
|
||||
history_mod.HISTORY_FILE.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"days": {
|
||||
"2026-08-28": {"seconds": 10, "games": {"ok": 5, "bad": "x"}},
|
||||
"2026-08-27": {"seconds": True},
|
||||
"2026-08-26": {"games": {}},
|
||||
"2026-08-25": ["nope"],
|
||||
},
|
||||
"labels": {"ok": "Fine", "bad": 7},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
days = {d.day: d.games for d in load_history()}
|
||||
assert days == {"2026-08-28": {"ok": 5.0}}
|
||||
assert history_mod.load_labels() == {"ok": "Fine"}
|
||||
|
||||
def test_non_mapping_games_is_ignored(self) -> None:
|
||||
history_mod.HISTORY_FILE.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": 1,
|
||||
"days": {"2026-08-28": {"seconds": 5, "games": 3}},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert [d.games for d in load_history()] == [{}]
|
||||
|
||||
def test_non_mapping_labels_is_ignored(self) -> None:
|
||||
history_mod.HISTORY_FILE.write_text(
|
||||
json.dumps({"schema_version": 1, "days": {}, "labels": "nope"}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert history_mod.load_labels() == {}
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import os
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
from steam_backlog_enforcer._playtime_notify import (
|
||||
_humanise,
|
||||
@ -12,13 +11,10 @@ from steam_backlog_enforcer._playtime_procs import (
|
||||
_match_cmdline,
|
||||
get_pids_by_cmdline_names,
|
||||
process_name,
|
||||
qualifying_pids,
|
||||
)
|
||||
from steam_backlog_enforcer._playtime_state import (
|
||||
PlaytimeState,
|
||||
rules_for,
|
||||
)
|
||||
from steam_backlog_enforcer.config import Config
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
@ -109,67 +105,6 @@ class TestGetPidsByCmdlineNames:
|
||||
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {}
|
||||
|
||||
|
||||
class TestQualifyingPids:
|
||||
def test_steam_games_only_when_launchers_are_off(self) -> None:
|
||||
rules = rules_for(Config(count_launcher_processes=False), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={5: 440, 6: 0},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names"
|
||||
) as mock_comm,
|
||||
):
|
||||
assert qualifying_pids(rules) == {5}
|
||||
mock_comm.assert_not_called()
|
||||
|
||||
def test_steam_client_appid_zero_does_not_count(self) -> None:
|
||||
"""Browsing the store is not gaming."""
|
||||
rules = rules_for(Config(count_launcher_processes=False), demo=False)
|
||||
with patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={9: 0},
|
||||
):
|
||||
assert qualifying_pids(rules) == set()
|
||||
|
||||
def test_unions_launchers_from_both_scanners(self) -> None:
|
||||
rules = rules_for(Config(count_launcher_processes=True), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={5: 440},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={7: "heroic"},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={8: "lutris"},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == {5, 7, 8}
|
||||
|
||||
def test_no_processes_at_all(self) -> None:
|
||||
rules = rules_for(Config(), demo=False)
|
||||
with (
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
|
||||
return_value={},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
|
||||
return_value={},
|
||||
),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
|
||||
return_value={},
|
||||
),
|
||||
):
|
||||
assert qualifying_pids(rules) == set()
|
||||
|
||||
|
||||
class TestHumanise:
|
||||
def test_seconds(self) -> None:
|
||||
assert _humanise(30) == "30 seconds"
|
||||
|
||||
@ -93,7 +93,7 @@ class TestPlaytimeTick:
|
||||
demo=False,
|
||||
)
|
||||
with (
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7}),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7: "app:440"}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
@ -108,7 +108,7 @@ class TestPlaytimeTick:
|
||||
)
|
||||
session = fake_session()
|
||||
with (
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7}),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7: "app:440"}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
@ -127,7 +127,7 @@ class TestPlaytimeTick:
|
||||
session = fake_session()
|
||||
with (
|
||||
patch(f"{PKG}.is_total_block_active", return_value=True),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7}),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7: "app:440"}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
@ -144,7 +144,7 @@ class TestPlaytimeTick:
|
||||
)
|
||||
with (
|
||||
patch(f"{PKG}.is_total_block_active", return_value=True),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7}),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7: "app:440"}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
@ -163,7 +163,7 @@ class TestPlaytimeTick:
|
||||
demo=False,
|
||||
)
|
||||
with (
|
||||
patch(f"{PKG}.qualifying_pids", return_value=set()),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
@ -182,7 +182,7 @@ class TestPlaytimeTick:
|
||||
demo=True,
|
||||
)
|
||||
with (
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7}),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7: "app:440"}),
|
||||
patch(
|
||||
"steam_backlog_enforcer._playtime_cutoff._kill_set", return_value={7}
|
||||
),
|
||||
@ -197,7 +197,7 @@ class TestPlaytimeTick:
|
||||
def test_demo_uses_a_separate_state_file(self, quiet_tick: dict) -> None:
|
||||
"""A demo run must never consume the real day's budget."""
|
||||
with (
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7}),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={7: "app:440"}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
@ -207,7 +207,7 @@ class TestPlaytimeTick:
|
||||
|
||||
def test_first_tick_writes_state(self, quiet_tick: dict) -> None:
|
||||
with (
|
||||
patch(f"{PKG}.qualifying_pids", return_value=set()),
|
||||
patch(f"{PKG}.qualifying_pids", return_value={}),
|
||||
patch(f"{PKG}.datetime") as mock_dt,
|
||||
):
|
||||
mock_dt.now.return_value = NOW
|
||||
|
||||
18
web/src/budgetSeries.test.ts
Normal file
18
web/src/budgetSeries.test.ts
Normal file
@ -0,0 +1,18 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { OTHER_KEY, SERIES_COLORS, UNATTRIBUTED_KEY, seriesClass } from './budgetSeries'
|
||||
|
||||
describe('seriesClass', () => {
|
||||
it('assigns ramp hues by legend position', () => {
|
||||
expect(seriesClass('app:1', 0)).toBe('seg cat-1')
|
||||
expect(seriesClass('app:2', 5)).toBe('seg cat-6')
|
||||
})
|
||||
|
||||
it('wraps rather than emitting a seventh hue that does not exist', () => {
|
||||
expect(seriesClass('app:3', SERIES_COLORS)).toBe('seg cat-1')
|
||||
})
|
||||
|
||||
it('keeps both sentinels off the ramp', () => {
|
||||
expect(seriesClass(UNATTRIBUTED_KEY, 0)).toBe('seg unattributed')
|
||||
expect(seriesClass(OTHER_KEY, 1)).toBe('seg other')
|
||||
})
|
||||
})
|
||||
29
web/src/budgetSeries.ts
Normal file
29
web/src/budgetSeries.ts
Normal file
@ -0,0 +1,29 @@
|
||||
/**
|
||||
* Colour assignment for the stacked budget chart.
|
||||
*
|
||||
* Kept out of the component because it is the one part with branches worth
|
||||
* testing directly, and because the rule it encodes is not local: `--cat-1..6`
|
||||
* come from the shared design system's categorical ramp, chosen under a
|
||||
* pairwise CIE dE >= 20 constraint including simulated colour-vision
|
||||
* deficiency. Six is the whole ramp -- a seventh distinguishable hue does not
|
||||
* exist in that band, which is why the backend folds the tail into `other`.
|
||||
*/
|
||||
|
||||
/** Games with no attribution, and the folded tail, sit outside the ramp. */
|
||||
export const UNATTRIBUTED_KEY = 'unattributed'
|
||||
export const OTHER_KEY = 'other'
|
||||
|
||||
/** Size of the categorical ramp; the backend caps series to match. */
|
||||
export const SERIES_COLORS = 6
|
||||
|
||||
/**
|
||||
* Return the CSS class for a legend entry at `index`.
|
||||
*
|
||||
* The two sentinels are deliberately neutral: they are absences of
|
||||
* information, and giving them a ramp hue would read as "another game".
|
||||
*/
|
||||
export function seriesClass(key: string, index: number): string {
|
||||
if (key === UNATTRIBUTED_KEY) return 'seg unattributed'
|
||||
if (key === OTHER_KEY) return 'seg other'
|
||||
return `seg cat-${(index % SERIES_COLORS) + 1}`
|
||||
}
|
||||
@ -1,19 +1,33 @@
|
||||
import { render, screen } from '@testing-library/react'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { BudgetHistoryDay, BudgetLegendEntry } from '../types'
|
||||
import { BudgetHistoryChart } from './BudgetHistoryChart'
|
||||
|
||||
const LEGEND: BudgetLegendEntry[] = [
|
||||
{ key: 'app:475150', label: 'Titan Quest' },
|
||||
{ key: 'proc:osu-lazer', label: 'osu!lazer' },
|
||||
{ key: 'unattributed', label: 'Unattributed' },
|
||||
]
|
||||
|
||||
function day(over: Partial<BudgetHistoryDay> = {}): BudgetHistoryDay {
|
||||
return {
|
||||
day: '2026-08-28',
|
||||
seconds: 14400,
|
||||
segments: [{ key: 'app:475150', seconds: 14400 }],
|
||||
...over,
|
||||
}
|
||||
}
|
||||
|
||||
describe('BudgetHistoryChart', () => {
|
||||
it('explains itself when nothing has been recorded', () => {
|
||||
render(<BudgetHistoryChart days={[]} budgetSeconds={28800} />)
|
||||
render(<BudgetHistoryChart days={[]} legend={[]} budgetSeconds={28800} />)
|
||||
expect(screen.getByText(/No budget history recorded yet/)).toBeInTheDocument()
|
||||
expect(document.querySelector('svg')).toBeNull()
|
||||
})
|
||||
|
||||
it('draws a single day without filling the plot', () => {
|
||||
render(
|
||||
<BudgetHistoryChart days={[{ day: '2026-08-28', seconds: 14400 }]} budgetSeconds={28800} />,
|
||||
)
|
||||
const bars = document.querySelectorAll('rect.bar')
|
||||
render(<BudgetHistoryChart days={[day()]} legend={LEGEND} budgetSeconds={28800} />)
|
||||
const bars = document.querySelectorAll('rect.seg')
|
||||
expect(bars).toHaveLength(1)
|
||||
// Scaled against the budget, half a budget is half the plot height.
|
||||
expect(Number(bars[0].getAttribute('height'))).toBeLessThan(
|
||||
@ -22,22 +36,79 @@ describe('BudgetHistoryChart', () => {
|
||||
expect(screen.getByText('08-28')).toBeInTheDocument()
|
||||
})
|
||||
|
||||
it('marks a day that reached the budget', () => {
|
||||
it('stacks one band per game and colours it by legend position', () => {
|
||||
render(
|
||||
<BudgetHistoryChart
|
||||
days={[
|
||||
{ day: '2026-08-27', seconds: 3600 },
|
||||
{ day: '2026-08-28', seconds: 28800 },
|
||||
day({
|
||||
seconds: 21600,
|
||||
segments: [
|
||||
{ key: 'app:475150', seconds: 14400 },
|
||||
{ key: 'proc:osu-lazer', seconds: 7200 },
|
||||
],
|
||||
}),
|
||||
]}
|
||||
legend={LEGEND}
|
||||
budgetSeconds={28800}
|
||||
/>,
|
||||
)
|
||||
expect(document.querySelectorAll('rect.bar')).toHaveLength(2)
|
||||
expect(document.querySelectorAll('rect.bar.over')).toHaveLength(1)
|
||||
expect(document.querySelectorAll('rect.seg')).toHaveLength(2)
|
||||
expect(document.querySelector('rect.cat-1')).not.toBeNull()
|
||||
expect(document.querySelector('rect.cat-2')).not.toBeNull()
|
||||
// Bands abut: the second starts exactly where the first ends.
|
||||
const [first, second] = document.querySelectorAll('rect.seg')
|
||||
const firstTop = Number(first.getAttribute('y'))
|
||||
const secondBottom =
|
||||
Number(second.getAttribute('y')) + Number(second.getAttribute('height'))
|
||||
expect(secondBottom).toBeCloseTo(firstTop, 5)
|
||||
})
|
||||
|
||||
it('survives a zero budget and zero seconds', () => {
|
||||
render(<BudgetHistoryChart days={[{ day: '2026-08-28', seconds: 0 }]} budgetSeconds={0} />)
|
||||
expect(document.querySelectorAll('rect')).toHaveLength(1)
|
||||
it('renders an unattributed day in its own neutral band', () => {
|
||||
render(
|
||||
<BudgetHistoryChart
|
||||
days={[
|
||||
day({
|
||||
day: '2026-08-24',
|
||||
seconds: 12000,
|
||||
segments: [{ key: 'unattributed', seconds: 12000 }],
|
||||
}),
|
||||
]}
|
||||
legend={LEGEND}
|
||||
budgetSeconds={28800}
|
||||
/>,
|
||||
)
|
||||
expect(document.querySelectorAll('rect.seg.unattributed')).toHaveLength(1)
|
||||
// No ramp hue: an absence of information must not read as another game.
|
||||
expect(document.querySelector('rect.cat-1')).toBeNull()
|
||||
})
|
||||
|
||||
it('labels each band with its game in the tooltip and the legend', () => {
|
||||
render(<BudgetHistoryChart days={[day()]} legend={LEGEND} budgetSeconds={28800} />)
|
||||
expect(document.querySelector('rect.seg title')?.textContent).toContain('Titan Quest')
|
||||
expect(document.querySelectorAll('.chart-legend li')).toHaveLength(LEGEND.length)
|
||||
})
|
||||
|
||||
it('falls back to the raw key when the legend omits it', () => {
|
||||
render(
|
||||
<BudgetHistoryChart
|
||||
days={[day({ segments: [{ key: 'app:999', seconds: 3600 }] })]}
|
||||
legend={LEGEND}
|
||||
budgetSeconds={28800}
|
||||
/>,
|
||||
)
|
||||
expect(document.querySelector('rect.seg title')?.textContent).toContain('app:999')
|
||||
})
|
||||
|
||||
it('survives a zero budget and a day that billed nothing', () => {
|
||||
render(
|
||||
<BudgetHistoryChart
|
||||
days={[day({ seconds: 0, segments: [] })]}
|
||||
legend={[]}
|
||||
budgetSeconds={0}
|
||||
/>,
|
||||
)
|
||||
// A day with no billed time draws no bands, but the axes still render.
|
||||
expect(document.querySelectorAll('rect.seg')).toHaveLength(0)
|
||||
expect(document.querySelector('svg')).not.toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
@ -1,8 +1,10 @@
|
||||
import { seriesClass } from '../budgetSeries'
|
||||
import { fmtDuration } from '../format'
|
||||
import type { BudgetHistoryDay } from '../types'
|
||||
import type { BudgetHistoryDay, BudgetLegendEntry } from '../types'
|
||||
|
||||
interface Props {
|
||||
days: BudgetHistoryDay[]
|
||||
legend: BudgetLegendEntry[]
|
||||
budgetSeconds: number
|
||||
}
|
||||
|
||||
@ -10,7 +12,7 @@ const W = 820
|
||||
const H = 220
|
||||
const PAD = { top: 16, right: 20, bottom: 36, left: 56 }
|
||||
|
||||
export function BudgetHistoryChart({ days, budgetSeconds }: Props) {
|
||||
export function BudgetHistoryChart({ days, legend, budgetSeconds }: Props) {
|
||||
if (days.length === 0) {
|
||||
return (
|
||||
<div className="chart">
|
||||
@ -34,6 +36,8 @@ export function BudgetHistoryChart({ days, budgetSeconds }: Props) {
|
||||
const barW = slot * 0.7
|
||||
|
||||
const yTicks = [0, 0.5, 1].map((f) => f * maxSeconds)
|
||||
const labelFor = new Map(legend.map((e) => [e.key, e.label]))
|
||||
const indexFor = new Map(legend.map((e, i) => [e.key, i]))
|
||||
|
||||
return (
|
||||
<div className="chart">
|
||||
@ -54,18 +58,31 @@ export function BudgetHistoryChart({ days, budgetSeconds }: Props) {
|
||||
y2={sy(budgetSeconds)}
|
||||
className="budget-line"
|
||||
/>
|
||||
{days.map((d, i) => (
|
||||
<rect
|
||||
key={d.day}
|
||||
x={PAD.left + i * slot + (slot - barW) / 2}
|
||||
y={sy(d.seconds)}
|
||||
width={barW}
|
||||
height={PAD.top + plotH - sy(d.seconds)}
|
||||
className={d.seconds >= budgetSeconds ? 'bar over' : 'bar'}
|
||||
>
|
||||
<title>{`${d.day}: ${fmtDuration(d.seconds)}`}</title>
|
||||
</rect>
|
||||
))}
|
||||
{days.map((d, i) => {
|
||||
// Bands stack upwards from the axis in legend order, so a game keeps
|
||||
// the same colour and the same relative position in every bar.
|
||||
let base = 0
|
||||
return d.segments.map((seg) => {
|
||||
const top = base + seg.seconds
|
||||
const y = sy(top)
|
||||
const height = sy(base) - y
|
||||
base = top
|
||||
return (
|
||||
<rect
|
||||
key={`${d.day}:${seg.key}`}
|
||||
x={PAD.left + i * slot + (slot - barW) / 2}
|
||||
y={y}
|
||||
width={barW}
|
||||
height={height}
|
||||
className={seriesClass(seg.key, indexFor.get(seg.key) ?? 0)}
|
||||
>
|
||||
<title>
|
||||
{`${d.day} · ${labelFor.get(seg.key) ?? seg.key}: ${fmtDuration(seg.seconds)}`}
|
||||
</title>
|
||||
</rect>
|
||||
)
|
||||
})
|
||||
})}
|
||||
{days.map((d, i) => (
|
||||
<text
|
||||
key={`x${d.day}`}
|
||||
@ -77,9 +94,17 @@ export function BudgetHistoryChart({ days, budgetSeconds }: Props) {
|
||||
</text>
|
||||
))}
|
||||
</svg>
|
||||
<ul className="chart-legend">
|
||||
{legend.map((entry, i) => (
|
||||
<li key={entry.key}>
|
||||
<span className={`swatch ${seriesClass(entry.key, i)}`} aria-hidden="true" />
|
||||
{entry.label}
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
<p className="hint">
|
||||
Billed gaming time per day. The dashed line is the daily budget; today's bar
|
||||
can still move.
|
||||
Billed gaming time per day, split by game. The dashed line is the daily budget;
|
||||
today's bar can still move.
|
||||
</p>
|
||||
</div>
|
||||
)
|
||||
|
||||
@ -7,7 +7,10 @@ describe('BudgetLiveCard', () => {
|
||||
it('reports an engaged session with its game', () => {
|
||||
render(<BudgetLiveCard session={makeBudgetSession()} />)
|
||||
expect(screen.getByText(/Billing — you are playing/)).toBeInTheDocument()
|
||||
expect(screen.getByText('Hollow Knight')).toBeInTheDocument()
|
||||
expect(screen.getByText(/Assigned game/)).toBeInTheDocument()
|
||||
// Engaged, so the credited game is described in the present tense.
|
||||
expect(screen.getByText(/Billing to/)).toBeInTheDocument()
|
||||
expect(screen.getAllByText('Hollow Knight')).toHaveLength(2)
|
||||
expect(screen.getByText(/3 qualifying processes/)).toBeInTheDocument()
|
||||
expect(screen.getByText(/Idle 2s/)).toBeInTheDocument()
|
||||
expect(screen.getByText(/Screen held: no/)).toBeInTheDocument()
|
||||
@ -59,7 +62,9 @@ describe('BudgetLiveCard', () => {
|
||||
})
|
||||
|
||||
it('handles a qualifying session with no assigned game name', () => {
|
||||
render(<BudgetLiveCard session={makeBudgetSession({ game_name: '' })} />)
|
||||
render(
|
||||
<BudgetLiveCard session={makeBudgetSession({ game_name: '', billing_label: '' })} />,
|
||||
)
|
||||
expect(screen.getByText(/3 qualifying processes/)).toBeInTheDocument()
|
||||
expect(screen.queryByText('Hollow Knight')).toBeNull()
|
||||
})
|
||||
|
||||
@ -49,6 +49,16 @@ export function BudgetLiveCard({ session }: Props) {
|
||||
{session.qualifying_count} qualifying processes
|
||||
</p>
|
||||
)}
|
||||
{session.billing_label && (
|
||||
<p className="hint">
|
||||
{/* The assignment above is what the enforcer told you to play; this
|
||||
is what the budget charged. A counted non-Steam game is never the
|
||||
assignment, so the two legitimately differ. The key is the *last*
|
||||
one credited, so it must not claim to be live while paused. */}
|
||||
{session.state === 'engaged' ? 'Billing to ' : 'Last billed to '}
|
||||
<strong>{session.billing_label}</strong>
|
||||
</p>
|
||||
)}
|
||||
{session.causes.length > 0 && (
|
||||
<p className="hint">Paused because {describe(session.causes, CAUSE_LABELS).join(', ')}.</p>
|
||||
)}
|
||||
|
||||
@ -40,10 +40,14 @@ export function BudgetPanel({ demo }: Props) {
|
||||
<p className="error">{data.error ?? 'Budget state unavailable.'}</p>
|
||||
</section>
|
||||
) : (
|
||||
<BudgetTodayCard today={data.today} maskedCount={data.rules.masked_launchers.length} />
|
||||
<BudgetTodayCard today={data.today} legend={data.legend} maskedCount={data.rules.masked_launchers.length} />
|
||||
)}
|
||||
<BudgetLiveCard session={data.session} />
|
||||
<BudgetHistoryChart days={data.history} budgetSeconds={data.rules.budget_seconds} />
|
||||
<BudgetHistoryChart
|
||||
days={data.history}
|
||||
legend={data.legend}
|
||||
budgetSeconds={data.rules.budget_seconds}
|
||||
/>
|
||||
<BudgetRulesCard rules={data.rules} />
|
||||
</main>
|
||||
)
|
||||
|
||||
@ -5,7 +5,7 @@ import { BudgetTodayCard } from './BudgetTodayCard'
|
||||
|
||||
describe('BudgetTodayCard', () => {
|
||||
it('shows remaining time, usage and the next warning', () => {
|
||||
render(<BudgetTodayCard today={makeBudgetToday()} maskedCount={0} />)
|
||||
render(<BudgetTodayCard today={makeBudgetToday()} legend={[]} maskedCount={0} />)
|
||||
expect(screen.getByText('7h 0m')).toBeInTheDocument()
|
||||
expect(screen.getByText(/1h 0m of 8h 0m used/)).toBeInTheDocument()
|
||||
expect(screen.getByText(/Next warning at 1h 0m remaining/)).toBeInTheDocument()
|
||||
@ -16,6 +16,8 @@ describe('BudgetTodayCard', () => {
|
||||
render(
|
||||
<BudgetTodayCard
|
||||
today={makeBudgetToday({ seconds_remaining: 900, warned_seconds: [3600, 1800] })}
|
||||
legend={[]}
|
||||
|
||||
maskedCount={0}
|
||||
/>,
|
||||
)
|
||||
@ -29,6 +31,8 @@ describe('BudgetTodayCard', () => {
|
||||
next_warning_seconds: null,
|
||||
warned_seconds: [3600, 1800, 600, 300],
|
||||
})}
|
||||
legend={[]}
|
||||
|
||||
maskedCount={0}
|
||||
/>,
|
||||
)
|
||||
@ -47,13 +51,13 @@ describe('BudgetTodayCard', () => {
|
||||
})
|
||||
|
||||
it('shows the cutoff badge when blocked', () => {
|
||||
render(<BudgetTodayCard today={blockedToday()} maskedCount={2} />)
|
||||
render(<BudgetTodayCard today={blockedToday()} legend={[]} maskedCount={2} />)
|
||||
expect(screen.getByText(/Cutoff engaged — launchers are masked/)).toBeInTheDocument()
|
||||
expect(document.querySelector('.budget-meter--blocked')).not.toBeNull()
|
||||
})
|
||||
|
||||
it('does not claim a mask that is not in place', () => {
|
||||
render(<BudgetTodayCard today={blockedToday()} maskedCount={0} />)
|
||||
render(<BudgetTodayCard today={blockedToday()} legend={[]} maskedCount={0} />)
|
||||
expect(
|
||||
screen.getByText(/Cutoff engaged — launchers are not currently masked/),
|
||||
).toBeInTheDocument()
|
||||
|
||||
@ -1,8 +1,10 @@
|
||||
import { seriesClass } from '../budgetSeries'
|
||||
import { fmtDuration } from '../format'
|
||||
import type { BudgetToday } from '../types'
|
||||
import type { BudgetLegendEntry, BudgetToday } from '../types'
|
||||
|
||||
interface Props {
|
||||
today: BudgetToday
|
||||
legend: BudgetLegendEntry[]
|
||||
maskedCount: number
|
||||
}
|
||||
|
||||
@ -18,7 +20,12 @@ function meterState(today: BudgetToday): string {
|
||||
return today.warned_seconds.length > 0 ? 'warn' : 'ok'
|
||||
}
|
||||
|
||||
export function BudgetTodayCard({ today, maskedCount }: Props) {
|
||||
export function BudgetTodayCard({ today, legend, maskedCount }: Props) {
|
||||
// Colour by the chart's legend position so a game reads the same in both
|
||||
// places; today can hold games the capped 14-day legend left out, and those
|
||||
// fall back to their own row order.
|
||||
const indexFor = new Map(legend.map((e, i) => [e.key, i]))
|
||||
|
||||
return (
|
||||
<section className="budget-today">
|
||||
<h2>Today</h2>
|
||||
@ -55,6 +62,26 @@ export function BudgetTodayCard({ today, maskedCount }: Props) {
|
||||
`Next warning at ${fmtDuration(today.next_warning_seconds)} remaining.`
|
||||
)}
|
||||
</p>
|
||||
{today.games.length > 0 && (
|
||||
<>
|
||||
<h3 className="budget-subhead">Today by game</h3>
|
||||
<ul className="budget-games">
|
||||
{today.games.map((game, i) => (
|
||||
<li key={game.key}>
|
||||
<span
|
||||
className={`swatch ${seriesClass(game.key, indexFor.get(game.key) ?? i)}`}
|
||||
aria-hidden="true"
|
||||
/>
|
||||
<span className="budget-game-name">{game.label}</span>
|
||||
<span className="budget-game-time">{fmtDuration(game.seconds)}</span>
|
||||
<span className="budget-game-share">
|
||||
{Math.round(game.fraction * 100)}%
|
||||
</span>
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
</>
|
||||
)}
|
||||
</section>
|
||||
)
|
||||
}
|
||||
|
||||
@ -14,6 +14,13 @@
|
||||
--on-fill: #211d1b;
|
||||
--accent-bg: rgba(184, 134, 46, 0.12);
|
||||
--accent-border: rgba(184, 134, 46, 0.4);
|
||||
--cat-1: #c57293;
|
||||
--cat-2: #398fc0;
|
||||
--cat-3: #c85a32;
|
||||
--cat-4: #228736;
|
||||
--cat-5: #8d58bb;
|
||||
--cat-6: #686d2c;
|
||||
--cat-none: #5d5551;
|
||||
--radius-sm: 8px;
|
||||
--radius-md: 12px;
|
||||
--sp-1: 4px;
|
||||
@ -632,14 +639,114 @@ td.proton {
|
||||
.budget-rules-list dd {
|
||||
margin: 0;
|
||||
}
|
||||
.chart-svg .bar {
|
||||
fill: var(--accent);
|
||||
.chart-svg .seg.cat-1 {
|
||||
fill: var(--cat-1);
|
||||
}
|
||||
.chart-svg .bar.over {
|
||||
fill: var(--danger);
|
||||
.chart-svg .seg.cat-2 {
|
||||
fill: var(--cat-2);
|
||||
}
|
||||
.chart-svg .seg.cat-3 {
|
||||
fill: var(--cat-3);
|
||||
}
|
||||
.chart-svg .seg.cat-4 {
|
||||
fill: var(--cat-4);
|
||||
}
|
||||
.chart-svg .seg.cat-5 {
|
||||
fill: var(--cat-5);
|
||||
}
|
||||
.chart-svg .seg.cat-6 {
|
||||
fill: var(--cat-6);
|
||||
}
|
||||
.chart-svg .seg.other {
|
||||
fill: var(--muted);
|
||||
}
|
||||
.chart-svg .seg.unattributed {
|
||||
fill: var(--cat-none);
|
||||
}
|
||||
.chart-svg .budget-line {
|
||||
stroke: var(--muted);
|
||||
stroke-width: 1;
|
||||
stroke-dasharray: 4 4;
|
||||
}
|
||||
|
||||
/* --- Per-game breakdown: legend, swatches, today's list --- */
|
||||
.chart-legend {
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: var(--sp-2) var(--sp-3);
|
||||
margin: var(--sp-2) 0 0;
|
||||
padding: 0;
|
||||
list-style: none;
|
||||
color: var(--muted);
|
||||
font-size: 0.875rem;
|
||||
}
|
||||
.chart-legend li {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: var(--sp-1);
|
||||
}
|
||||
.swatch {
|
||||
display: inline-block;
|
||||
width: 12px;
|
||||
height: 12px;
|
||||
border-radius: 3px;
|
||||
background: var(--cat-none);
|
||||
flex: none;
|
||||
}
|
||||
.swatch.cat-1 {
|
||||
background: var(--cat-1);
|
||||
}
|
||||
.swatch.cat-2 {
|
||||
background: var(--cat-2);
|
||||
}
|
||||
.swatch.cat-3 {
|
||||
background: var(--cat-3);
|
||||
}
|
||||
.swatch.cat-4 {
|
||||
background: var(--cat-4);
|
||||
}
|
||||
.swatch.cat-5 {
|
||||
background: var(--cat-5);
|
||||
}
|
||||
.swatch.cat-6 {
|
||||
background: var(--cat-6);
|
||||
}
|
||||
.swatch.other {
|
||||
background: var(--muted);
|
||||
}
|
||||
.swatch.unattributed {
|
||||
background: var(--cat-none);
|
||||
}
|
||||
.budget-subhead {
|
||||
margin: var(--sp-3) 0 var(--sp-2);
|
||||
font-size: 0.95rem;
|
||||
color: var(--heading);
|
||||
}
|
||||
.budget-games {
|
||||
margin: 0;
|
||||
padding: 0;
|
||||
list-style: none;
|
||||
display: grid;
|
||||
gap: var(--sp-1);
|
||||
}
|
||||
.budget-games li {
|
||||
display: grid;
|
||||
grid-template-columns: 12px 1fr auto auto;
|
||||
align-items: center;
|
||||
gap: var(--sp-2);
|
||||
}
|
||||
.budget-game-name {
|
||||
color: var(--text);
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
}
|
||||
.budget-game-time {
|
||||
font-family: var(--mono);
|
||||
color: var(--text);
|
||||
}
|
||||
.budget-game-share {
|
||||
color: var(--muted);
|
||||
min-width: 3ch;
|
||||
text-align: right;
|
||||
}
|
||||
|
||||
@ -114,6 +114,9 @@ export function makeBudgetToday(over: Partial<BudgetToday> = {}): BudgetToday {
|
||||
blocked_at: 0,
|
||||
next_warning_seconds: 3600,
|
||||
warned_seconds: [],
|
||||
games: [
|
||||
{ key: 'app:367520', label: 'Hollow Knight', seconds: 3600, fraction: 1 },
|
||||
],
|
||||
...over,
|
||||
}
|
||||
}
|
||||
@ -128,6 +131,7 @@ export function makeBudgetSession(over: Partial<BudgetSession> = {}): BudgetSess
|
||||
idle_seconds: 2,
|
||||
screen_held: false,
|
||||
game_name: 'Hollow Knight',
|
||||
billing_label: 'Hollow Knight',
|
||||
qualifying_count: 3,
|
||||
processes: [{ pid: 42, name: 'hollow_knight' }],
|
||||
...over,
|
||||
@ -157,7 +161,14 @@ export function makeBudget(over: Partial<BudgetSnapshot> = {}): BudgetSnapshot {
|
||||
error: null,
|
||||
today: makeBudgetToday(),
|
||||
session: makeBudgetSession(),
|
||||
history: [{ day: '2026-08-28', seconds: 3600 }],
|
||||
history: [
|
||||
{
|
||||
day: '2026-08-28',
|
||||
seconds: 3600,
|
||||
segments: [{ key: 'app:367520', seconds: 3600 }],
|
||||
},
|
||||
],
|
||||
legend: [{ key: 'app:367520', label: 'Hollow Knight' }],
|
||||
rules: makeBudgetRules(),
|
||||
...over,
|
||||
}
|
||||
|
||||
@ -114,6 +114,19 @@ export interface BudgetToday {
|
||||
/** Seconds-remaining at which the next warning fires; null when none left. */
|
||||
next_warning_seconds: number | null
|
||||
warned_seconds: number[]
|
||||
/** Every game billed today, largest first, remainder last. Uncapped. */
|
||||
games: BudgetGameSlice[]
|
||||
}
|
||||
|
||||
/** One game's share of a day's billed time. */
|
||||
export interface BudgetGameSlice {
|
||||
/** Stable attribution key: `app:<id>`, `proc:<id>`, `launcher:<name>`,
|
||||
* or the computed `unattributed`/`other` sentinels. */
|
||||
key: string
|
||||
label: string
|
||||
seconds: number
|
||||
/** 0..1 of the day's billed total. */
|
||||
fraction: number
|
||||
}
|
||||
|
||||
export interface BudgetProcess {
|
||||
@ -131,6 +144,9 @@ export interface BudgetSession {
|
||||
idle_seconds: number | null
|
||||
screen_held: boolean | null
|
||||
game_name: string
|
||||
/** Game the budget is actually charging; differs from `game_name`, which
|
||||
* is the backlog assignment. Empty when nothing has billed yet. */
|
||||
billing_label: string
|
||||
qualifying_count: number
|
||||
processes: BudgetProcess[]
|
||||
}
|
||||
@ -138,6 +154,20 @@ export interface BudgetSession {
|
||||
export interface BudgetHistoryDay {
|
||||
day: string
|
||||
seconds: number
|
||||
/** Stacked bands, already ordered to match `BudgetSnapshot.legend` and
|
||||
* already capped at six games plus `other`/`unattributed` by the backend. */
|
||||
segments: BudgetHistorySegment[]
|
||||
}
|
||||
|
||||
export interface BudgetHistorySegment {
|
||||
key: string
|
||||
seconds: number
|
||||
}
|
||||
|
||||
/** Render order and labels for the stacked bands, shared by every day. */
|
||||
export interface BudgetLegendEntry {
|
||||
key: string
|
||||
label: string
|
||||
}
|
||||
|
||||
export interface BudgetRules {
|
||||
@ -161,6 +191,7 @@ export interface BudgetSnapshot {
|
||||
today: BudgetToday | null
|
||||
session: BudgetSession
|
||||
history: BudgetHistoryDay[]
|
||||
legend: BudgetLegendEntry[]
|
||||
rules: BudgetRules
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user