kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 17:47:24 +01:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

focus_project: registry, stands_down, activity

Browse Source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWYcmik6APnHfc9gEVy
This commit is contained in:
Krzysztof Rudnicki 2026-11-02 08:13:52 +01:00
parent 1e96563522
commit f99b77af89
21 changed files with 1272 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
.github/workflows/focus-project-precommit.yml vendored Normal file
View File

@ -0,0 +1,73 @@
name: focus-project pre-commit
on:
push:
branches: [main]
# The workflow's own file counts as focus_project: editing only this file would
# otherwise trigger nothing, so a fix to the job itself could never prove
# itself.
paths: ['focus_project/**', '.github/workflows/focus-project-precommit.yml']
pull_request:
branches: [main]
paths: ['focus_project/**', '.github/workflows/focus-project-precommit.yml']
jobs:
pre-commit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.x"
# Only pre-commit itself. Every linting hook runs in its own isolated
# env -- ruff, mypy, pylint, bandit and codespell each come from their
# own repo+rev, and the local hooks need nothing beyond bash and
# python3 -- so installing the full requirements.txt spent 9s a run on
# packages no pre-commit-stage hook imports. Verified by running all
# three configs green in a venv holding pre-commit and nothing else.
# The pin is read out of requirements.txt rather than repeated here, so
# it cannot drift from what the dependency-freshness gate governs.
- name: Install pre-commit
run: |
pip install "$(grep '^pre-commit==' focus_project/requirements.txt)"
# Keyed on THIS package's config. pre-commit/action@v3.0.1 caches on
# hashFiles('.pre-commit-config.yaml') -- the repo ROOT's -- so all three
# package jobs shared one key, raced to overwrite it with hook sets built
# from four different configs, and editing a per-package config did not
# invalidate it. The cache was therefore silently wrong rather than
# missing, and the hooks reinstalled from scratch on every run (38-49s).
- uses: actions/cache@v4
with:
path: ~/.cache/pre-commit
key: >-
pre-commit-focus-project-${{ env.pythonLocation }}-${{
hashFiles('focus_project/.pre-commit-config.yaml') }}
# Prefix fallback, or this fix would evaporate on the very
# event that motivated it: bumping a hook rev changes the
# exact key, and a total miss means reinstalling every hook
# again. pre-commit's store is keyed internally by hook
# repo+rev, so a superseded entry is ignored disk rather
# than a wrong hook, and cache still re-saves under the new
# exact key after a prefix-only hit.
restore-keys: pre-commit-focus-project-${{ env.pythonLocation }}-
# mypy re-analyses the stdlib and every third-party stub from scratch
# without this, which measured 10s of the hook step in CI against 0.25s
# warm locally -- a 19MB cache dir does the whole difference. Keyed on
# the config, not the sources: mypy's own cache metadata records the
# interpreter, the flag set and a hash per source file, so it re-checks
# what changed and cannot pass stale code. Deliberately NOT keyed on
# github.sha -- that re-saves 19MB x3 every commit and would evict the
# 68MB hook caches under the repo's 10GB LRU budget. A fixed key keeps
# the expensive part, the dependency analysis, which is what costs 10s.
- uses: actions/cache@v4
with:
path: .mypy_cache
key: >-
mypy-focus-project-${{ env.pythonLocation }}-${{
hashFiles('focus_project/.pre-commit-config.yaml') }}
restore-keys: mypy-focus-project-${{ env.pythonLocation }}-
- name: Run the hooks
run: >-
pre-commit run --config focus_project/.pre-commit-config.yaml
--all-files --show-diff-on-failure

34
.github/workflows/focus-project-tests.yml vendored Normal file
View File

@ -0,0 +1,34 @@
name: focus-project tests
on:
push:
branches: [main]
# The workflow's own file counts as focus_project: editing only this file would
# otherwise trigger nothing, so a fix to the job itself could never prove
# itself.
paths: ['focus_project/**', '.github/workflows/focus-project-tests.yml']
pull_request:
branches: [main]
paths: ['focus_project/**', '.github/workflows/focus-project-tests.yml']
jobs:
test:
runs-on: ubuntu-latest
defaults:
run:
working-directory: focus_project
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.x"
cache: pip
cache-dependency-path: focus_project/requirements.txt
- name: Install dependencies
run: pip install -r requirements.txt && pip install --no-deps -e .
- name: Run tests with coverage
run: python -m pytest -v

146
focus_project/.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,146 @@
# Pre-commit Configuration for focus_project
# Install: pre-commit install && pre-commit install --hook-type pre-push
# Run: pre-commit run --all-files
# Update: pre-commit autoupdate
# NOTE: this subproject lives inside the ~/src/utils monorepo, whose git root is
# one level up. pre-commit always runs hooks from the git root, so every path
# below is written relative to THAT, not to this directory, and the top-level
# `files` key keeps these hooks off the sibling subprojects. CI invokes this
# config exactly the same way:
# pre-commit run --config focus_project/.pre-commit-config.yaml --all-files
files: ^focus_project/
default_language_version:
python: python3
default_stages: [pre-commit]
fail_fast: false
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.6.0
hooks:
- id: trailing-whitespace
args: [--markdown-linebreak-ext=md]
- id: end-of-file-fixer
- id: check-yaml
- id: check-json
- id: check-toml
- id: check-added-large-files
args: [--maxkb=2000]
- id: check-merge-conflict
- id: detect-private-key
- id: debug-statements
- id: name-tests-test
args: [--pytest-test-first]
- id: check-ast
- id: mixed-line-ending
args: [--fix=lf]
- id: requirements-txt-fixer
- repo: local
hooks:
- id: file-length-cap
name: file length <= 250 lines
entry: bash scripts/check_file_length.sh
language: system
pass_filenames: true
- id: subproject-scope
name: subproject scope regex still matches
entry: scripts/check_subproject_scope.sh focus_project
language: script
# always_run so the top-level `files` filter cannot skip the one hook
# whose job is to notice that filter has stopped matching anything.
always_run: true
pass_filenames: false
- id: no-noqa
name: Block noqa comments
entry: '(?i)#\s*(noqa|type:\s*ignore)'
language: pygrep
types: [python]
- id: no-ruff-noqa
name: Block ruff noqa file-level comments
entry: '(?i)#\s*ruff:\s*noqa'
language: pygrep
types: [python]
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.17.2
hooks:
- id: ruff
args: [--fix, --unsafe-fixes, --exit-non-zero-on-fix, --show-fixes]
types_or: [python, pyi]
- id: ruff-format
types_or: [python, pyi]
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.13.0
hooks:
- id: mypy
args:
- --ignore-missing-imports
- --no-error-summary
- --disable-error-code=no-untyped-def
- --disable-error-code=no-untyped-call
- --disable-error-code=var-annotated
- --disable-error-code=no-any-unimported
- --disable-error-code=type-arg
- --disable-error-code=no-any-return
- --disable-error-code=misc
- --disable-error-code=unused-ignore
- --disable-error-code=unreachable
- --disable-error-code=assignment
- --disable-error-code=no-redef
- --disable-error-code=attr-defined
- --disable-error-code=arg-type
- --disable-error-code=union-attr
- --disable-error-code=call-overload
- --disable-error-code=return-value
- --disable-error-code=redundant-cast
- --disable-error-code=empty-body
- --disable-error-code=list-item
- repo: https://github.com/pylint-dev/pylint
rev: v4.0.8
hooks:
- id: pylint
args:
- --rcfile=focus_project/pyproject.toml
- --fail-under=10
- --jobs=4
additional_dependencies:
- pytest
# Kept in step with pyproject.toml, so pylint resolves the crdt_sync
# imports instead of scoring every one of them as import-error.
- crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync
- repo: https://github.com/PyCQA/bandit
rev: 1.9.4
hooks:
- id: bandit
args:
- -c
- focus_project/pyproject.toml
- --severity-level=high
- --confidence-level=medium
- --skip=B113
additional_dependencies: ["bandit[toml]"]
exclude: ^(focus_project/focus_project/tests/|.*test.*\.py$)
- repo: local
hooks:
- id: pytest-coverage
name: pytest with coverage enforcement
entry: scripts/run_subproject_tests.sh focus_project
language: script
types: [python]
pass_filenames: false
require_serial: true
stages: [pre-push]
- repo: https://github.com/codespell-project/codespell
rev: v2.3.0
hooks:
- id: codespell
args:
- --skip=*.json,*.lock,.git,__pycache__,.venv

17
focus_project/README.md Normal file
View File

@ -0,0 +1,17 @@
# focus_project
Time-boxed focus projects ("until Thursday I prepare for X"). The active
project switches chosen gates off and defines a daily-minutes gate.
```python
import focus_project as fp
fp.stands_down("leetcode", today) # gate in leetcode|book|reading|tutor; never raises
fp.active(today) # Project | None
fp.status(today) # dict for i3bar / GUIs
fp.record_activity("timer", "start")
```
Data: `$FOCUS_PROJECT_DATA` or `~/.local/share/focus_project/` (`projects.json`,
`activity.jsonl`). `minutes.py` is owned by the minutes/harvester agent.
Install: `./install.sh` (stdlib only, editable user install for `/usr/bin/python3`).

41
focus_project/focus_project/__init__.py Normal file
View File

@ -0,0 +1,41 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Time-boxed focus projects that replace chosen gates.
A gate's whole integration::
import focus_project
if focus_project.stands_down("leetcode", today):
... # this gate is off for the active project: no charge, debt or lock
``stands_down`` never raises: an unreadable registry reads as "not standing
down", so the failure mode is "the normal rules apply".
"""
from __future__ import annotations
from focus_project._api import (
active,
create_project,
end_project,
list_projects,
stands_down,
)
from focus_project._errors import RegistryError
from focus_project._model import Project
from focus_project.activity import record_activity
from focus_project.minutes import minutes_for
from focus_project.status import status
__all__ = [
"Project",
"RegistryError",
"active",
"create_project",
"end_project",
"list_projects",
"minutes_for",
"record_activity",
"stands_down",
"status",
]

127
focus_project/focus_project/_api.py Normal file
View File

@ -0,0 +1,127 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The registry API: active, stands_down, create/end/list."""
from __future__ import annotations
from dataclasses import replace
from datetime import date # noqa: TC003
import logging
import re
import time
from focus_project import _store
from focus_project._errors import RegistryError
from focus_project._model import GATES, Project, normalize_gate
_LOG = logging.getLogger(__name__)
_ID = re.compile(r"[a-z0-9][a-z0-9-]*")
def list_projects() -> list[Project]:
"""Every project in the registry, in file order."""
return _store.read_all()
def active(day: date) -> Project | None:
"""The project active on ``day``, or None. Raises on a corrupt registry."""
hits = [p for p in _store.read_all() if p.covers(day)]
if len(hits) > 1:
msg = f"projects overlap on {day}: {[p.id for p in hits]}"
raise RegistryError(msg)
return hits[0] if hits else None
def stands_down(gate: str, day: date) -> bool:
"""True iff the active project replaces ``gate``; never raises.
Fail-closed: an unreadable registry means the gate stays ON (logged),
never that a gate is silently switched off.
"""
try:
project = active(day)
except (RegistryError, OSError) as exc:
_LOG.warning(
"focus_project registry unreadable, gate %s stays on: %s", gate, exc
)
return False
return project is not None and project.replaces_gate(gate)
def _validate(project: Project) -> None:
if not _ID.fullmatch(project.id):
msg = f"id {project.id!r} must be a lowercase slug"
raise ValueError(msg)
if not project.name.strip():
msg = "name must not be blank"
raise ValueError(msg)
if project.end < project.start:
msg = "end is before start"
raise ValueError(msg)
if not project.replaces or not set(project.replaces) <= GATES:
msg = f"replaces must be a non-empty subset of gates {sorted(GATES)}"
raise ValueError(msg)
if project.daily_minutes <= 0:
msg = "daily_minutes must be positive"
raise ValueError(msg)
if not 0 <= project.unlock_minutes <= project.daily_minutes:
msg = "unlock_minutes must be within 0..daily_minutes"
raise ValueError(msg)
def create_project( # noqa: PLR0913
project_id: str,
*,
name: str,
start: date,
end: date,
replaces: list[str],
daily_minutes: int,
unlock_minutes: int,
subjects: list[str] | None = None,
site_patterns: list[str] | None = None,
) -> Project:
"""Add a project; ValueError on bad input, a duplicate id or an overlap."""
project = Project(
id=project_id,
name=name,
start=start,
end=end,
replaces=tuple(dict.fromkeys(normalize_gate(g) for g in replaces)),
daily_minutes=daily_minutes,
unlock_minutes=unlock_minutes,
subjects=tuple(subjects or ()),
site_patterns=tuple(site_patterns or ()),
created_ms=time.time_ns() // 1_000_000,
)
_validate(project)
with _store.locked():
existing = _store.read_all()
for other in existing:
if other.id == project.id:
msg = f"project {project.id!r} already exists"
raise ValueError(msg)
if project.start <= other.last_day and other.start <= project.last_day:
msg = f"project overlaps {other.id!r} ({other.start}..{other.last_day})"
raise ValueError(msg)
_store.write_all([*existing, project])
return project
def end_project(project_id: str, day: date) -> Project:
"""End early: ``day`` becomes the last active day (clamped to ``end``)."""
with _store.locked():
existing = _store.read_all()
for i, project in enumerate(existing):
if project.id != project_id:
continue
if project.ended_on is not None:
msg = f"project {project_id!r} already ended on {project.ended_on}"
raise ValueError(msg)
if day < project.start:
msg = f"cannot end {project_id!r} before its start {project.start}"
raise ValueError(msg)
ended = replace(project, ended_on=min(day, project.end))
existing[i] = ended
_store.write_all(existing)
return ended
raise KeyError(project_id)

8
focus_project/focus_project/_errors.py Normal file
View File

@ -0,0 +1,8 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Errors raised by the registry."""
from __future__ import annotations
class RegistryError(ValueError):
"""projects.json is unreadable, malformed or self-contradictory."""

86
focus_project/focus_project/_model.py Normal file
View File

@ -0,0 +1,86 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The :class:`Project` value and the gate vocabulary."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from typing import Any
from focus_project._errors import RegistryError
GATES = frozenset({"leetcode", "reading", "tutor"})
_ALIASES = {"book": "reading"}
def normalize_gate(gate: str) -> str:
"""Map the book/reading synonyms onto one name; others pass through."""
return _ALIASES.get(gate, gate)
@dataclass(frozen=True)
class Project:
"""One time-boxed focus project; at most one is active per day."""
id: str
name: str
start: date
end: date
replaces: tuple[str, ...]
daily_minutes: int
unlock_minutes: int
subjects: tuple[str, ...] = ()
site_patterns: tuple[str, ...] = ()
ended_on: date | None = None
created_ms: int = 0
@property
def last_day(self) -> date:
"""The last active day: the early-end day if set, else ``end``."""
return self.ended_on if self.ended_on is not None else self.end
def covers(self, day: date) -> bool:
"""Whether this project is active on ``day``."""
return self.start <= day <= self.last_day
def replaces_gate(self, gate: str) -> bool:
"""Whether this project switches ``gate`` off."""
return normalize_gate(gate) in self.replaces
def to_dict(self) -> dict[str, Any]:
"""The projects.json row."""
return {
"id": self.id,
"name": self.name,
"start": self.start.isoformat(),
"end": self.end.isoformat(),
"replaces": list(self.replaces),
"daily_minutes": self.daily_minutes,
"unlock_minutes": self.unlock_minutes,
"subjects": list(self.subjects),
"site_patterns": list(self.site_patterns),
"ended_on": self.ended_on.isoformat() if self.ended_on else None,
"created_ms": self.created_ms,
}
@classmethod
def from_dict(cls, row: Any) -> Project: # noqa: ANN401
"""Parse a registry row, or raise :class:`RegistryError`."""
try:
ended = row.get("ended_on")
return cls(
id=str(row["id"]),
name=str(row["name"]),
start=date.fromisoformat(row["start"]),
end=date.fromisoformat(row["end"]),
replaces=tuple(row["replaces"]),
daily_minutes=int(row["daily_minutes"]),
unlock_minutes=int(row["unlock_minutes"]),
subjects=tuple(row.get("subjects", ())),
site_patterns=tuple(row.get("site_patterns", ())),
ended_on=date.fromisoformat(ended) if ended else None,
created_ms=int(row.get("created_ms", 0)),
)
except (KeyError, TypeError, ValueError, AttributeError) as exc:
msg = f"bad project row {row!r}: {exc}"
raise RegistryError(msg) from exc

32
focus_project/focus_project/_paths.py Normal file
View File

@ -0,0 +1,32 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Where the data lives, resolved on every call so tests can redirect it."""
from __future__ import annotations
import os
from pathlib import Path
ENV_VAR = "FOCUS_PROJECT_DATA"
def data_dir() -> Path:
"""``$FOCUS_PROJECT_DATA`` or ``~/.local/share/focus_project``."""
override = os.environ.get(ENV_VAR)
if override:
return Path(override)
return Path.home() / ".local" / "share" / "focus_project"
def projects_path() -> Path:
"""The registry file; this package is its only writer."""
return data_dir() / "projects.json"
def activity_path() -> Path:
"""The append-only activity log."""
return data_dir() / "activity.jsonl"
def lock_path() -> Path:
"""The registry's writer lock."""
return data_dir() / "projects.lock"

61
focus_project/focus_project/_store.py Normal file
View File

@ -0,0 +1,61 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""projects.json: whole-file atomic writes under one flock (single writer)."""
from __future__ import annotations
from contextlib import contextmanager
import fcntl
import json
import os
from pathlib import Path
import tempfile
from typing import TYPE_CHECKING
from focus_project._errors import RegistryError
from focus_project._model import Project
from focus_project._paths import data_dir, lock_path, projects_path
if TYPE_CHECKING:
from collections.abc import Iterator
@contextmanager
def locked() -> Iterator[None]:
"""Hold the exclusive writer lock for a read-modify-write."""
data_dir().mkdir(parents=True, exist_ok=True)
with lock_path().open("a") as handle:
fcntl.flock(handle, fcntl.LOCK_EX)
yield
def read_all() -> list[Project]:
"""Every project; a missing file is empty, a bad one raises."""
try:
text = projects_path().read_text(encoding="utf-8")
except FileNotFoundError:
return []
try:
rows = json.loads(text)["projects"]
if not isinstance(rows, list):
msg = "'projects' is not a list"
raise TypeError(msg) # noqa: TRY301
except (ValueError, KeyError, TypeError) as exc:
msg = f"unreadable {projects_path()}: {exc}"
raise RegistryError(msg) from exc
return [Project.from_dict(row) for row in rows]
def write_all(projects: list[Project]) -> None:
"""Replace the file atomically: temp file in the same dir, fsync, rename."""
target = projects_path()
payload = json.dumps({"projects": [p.to_dict() for p in projects]}, indent=2)
fd, tmp = tempfile.mkstemp(dir=target.parent, prefix="projects.", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload + "\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp, target) # noqa: PTH105
except BaseException:
Path(tmp).unlink(missing_ok=True)
raise

58
focus_project/focus_project/activity.py Normal file
View File

@ -0,0 +1,58 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Append-only activity.jsonl; one flock'd, fsync'd line per event."""
from __future__ import annotations
from datetime import date
import fcntl
import json
import os
import time
from typing import Any
from focus_project import _api
from focus_project._paths import activity_path, data_dir
SOURCES = frozenset({"timer", "tutor", "sql_drills", "leetcode", "browser"})
KINDS = frozenset({"start", "checkin", "stop", "heartbeat"})
DEVICES = frozenset({"pc", "phone"})
def record_activity( # noqa: PLR0913, PLR0917
source: str,
kind: str,
ts: float | None = None,
project: str | None = None,
dev: str = "pc",
meta: dict[str, Any] | None = None,
) -> None:
"""Append one event. ``project`` defaults to the project active at ``ts``."""
for value, allowed, label in (
(source, SOURCES, "source"),
(kind, KINDS, "kind"),
(dev, DEVICES, "dev"),
):
if value not in allowed:
msg = f"unknown {label} {value!r}; expected one of {sorted(allowed)}"
raise ValueError(msg)
when = time.time() if ts is None else ts
if project is None:
found = _api.active(date.fromtimestamp(when)) # noqa: DTZ012
project = found.id if found else None
line = json.dumps(
{
"ts": when,
"source": source,
"kind": kind,
"project": project,
"dev": dev,
"meta": meta or {},
},
separators=(",", ":"),
)
data_dir().mkdir(parents=True, exist_ok=True)
with activity_path().open("a", encoding="utf-8") as handle:
fcntl.flock(handle, fcntl.LOCK_EX)
handle.write(line + "\n")
handle.flush()
os.fsync(handle.fileno())

15
focus_project/focus_project/minutes.py Normal file
View File

@ -0,0 +1,15 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Credited focus minutes per day. Owned by agent D: replace the body.
Contract: union of activity intervals on the local calendar day, capped at
the active project's daily_minutes. This stub returns 0 until D lands.
"""
from __future__ import annotations
from datetime import date # noqa: TC003
def minutes_for(day: date) -> int: # noqa: ARG001
"""Minutes of focus work credited on ``day`` (stub: always 0)."""
return 0

0
focus_project/focus_project/py.typed Normal file
View File

33
focus_project/focus_project/status.py Normal file
View File

@ -0,0 +1,33 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""status(day): the dict i3bar, daily-limits and the GUIs render."""
from __future__ import annotations
from datetime import date # noqa: TC003
from typing import Any
from focus_project import _api, minutes
def status(day: date) -> dict[str, Any]:
"""Progress of the project active on ``day``; ``project`` is None if none."""
project = _api.active(day)
if project is None:
return {
"project": None,
"name": None,
"minutes": 0,
"daily_minutes": None,
"unlock_minutes": None,
"ends": None,
"replaces": [],
}
return {
"project": project.id,
"name": project.name,
"minutes": minutes.minutes_for(day),
"daily_minutes": project.daily_minutes,
"unlock_minutes": project.unlock_minutes,
"ends": project.last_day.isoformat(),
"replaces": list(project.replaces),
}

1
focus_project/focus_project/tests/__init__.py Normal file
View File

@ -0,0 +1 @@
# Copyright (c) 2026 Krzysztof Rudnicki

27
focus_project/focus_project/tests/conftest.py Normal file
View File

@ -0,0 +1,27 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Redirect the data directory away from the real one, for every test.
Every path resolves through :func:`focus_project._paths.data_dir`, which
honours ``FOCUS_PROJECT_DATA``, so one env var moves all of them.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _redirect_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Point the data directory at ``tmp_path`` before any test runs."""
monkeypatch.setenv("FOCUS_PROJECT_DATA", str(tmp_path / "data"))
@pytest.fixture
def data(tmp_path: Path) -> Path:
"""The redirected data directory (may not exist yet)."""
return tmp_path / "data"

147
focus_project/focus_project/tests/test_activity_status.py Normal file
View File

@ -0,0 +1,147 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""record_activity, status, paths and the minutes stub."""
from __future__ import annotations
from datetime import date
import json
from pathlib import Path # noqa: TC003
import threading
import pytest
import focus_project as fp
from focus_project import minutes
from focus_project._paths import data_dir
D1 = date(2026, 11, 2)
def _project() -> None:
fp.create_project(
"hr",
name="HackerRank prep",
start=D1,
end=date(2026, 11, 5),
replaces=["leetcode"],
daily_minutes=120,
unlock_minutes=30,
)
def test_data_dir_env_and_default(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
assert data_dir() == tmp_path / "data"
monkeypatch.delenv("FOCUS_PROJECT_DATA")
monkeypatch.setenv("HOME", str(tmp_path / "home"))
assert data_dir() == tmp_path / "home" / ".local" / "share" / "focus_project"
def test_record_activity_line(data: Path) -> None:
_project()
fp.record_activity(
"timer", "start", ts=1_000.5, project="hr", dev="phone", meta={"k": 1}
)
row = json.loads((data / "activity.jsonl").read_text())
assert row == {
"ts": 1_000.5,
"source": "timer",
"kind": "start",
"project": "hr",
"dev": "phone",
"meta": {"k": 1},
}
def test_record_activity_defaults_now_and_project(data: Path) -> None:
today = date.today() # noqa: DTZ011
fp.create_project(
"now",
name="n",
start=today,
end=today,
replaces=["tutor"],
daily_minutes=60,
unlock_minutes=10,
)
fp.record_activity("tutor", "heartbeat")
fp.record_activity("tutor", "heartbeat", project=None)
rows = [json.loads(x) for x in (data / "activity.jsonl").read_text().splitlines()]
assert [r["project"] for r in rows] == ["now", "now"]
assert rows[0]["ts"] > 1_700_000_000
assert rows[0]["meta"] == {}
assert rows[0]["dev"] == "pc"
def test_record_activity_no_project_when_none_active(data: Path) -> None:
fp.record_activity("browser", "heartbeat", ts=1_000.0)
assert json.loads((data / "activity.jsonl").read_text())["project"] is None
@pytest.mark.parametrize(
"kw",
[
{"source": "x", "kind": "start"},
{"source": "timer", "kind": "x"},
{"source": "timer", "kind": "start", "dev": "tv"},
],
)
def test_record_activity_validation(kw: dict[str, str], data: Path) -> None:
with pytest.raises(ValueError, match="unknown"):
fp.record_activity(**kw) # type: ignore[arg-type]
assert not (data / "activity.jsonl").exists()
def test_record_activity_concurrent_appends(data: Path) -> None:
def work(n: int) -> None:
for i in range(20):
fp.record_activity(
"tutor", "heartbeat", ts=float(n * 100 + i), meta={"pad": "x" * 5000}
)
threads = [threading.Thread(target=work, args=(n,)) for n in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
lines = (data / "activity.jsonl").read_text().splitlines()
assert len(lines) == 160
assert all(json.loads(x)["source"] == "tutor" for x in lines)
def test_status_active_uses_minutes_module(monkeypatch: pytest.MonkeyPatch) -> None:
_project()
assert fp.status(D1)["minutes"] == 0 # the stub
monkeypatch.setattr(minutes, "minutes_for", lambda _day: 45)
st = fp.status(D1)
assert st == {
"project": "hr",
"name": "HackerRank prep",
"minutes": 45,
"daily_minutes": 120,
"unlock_minutes": 30,
"ends": "2026-11-05",
"replaces": ["leetcode"],
}
def test_status_inactive() -> None:
st = fp.status(D1)
assert st["project"] is None
assert st["minutes"] == 0
assert st["replaces"] == []
assert st["ends"] is None
assert set(st) == {
"project",
"name",
"minutes",
"daily_minutes",
"unlock_minutes",
"ends",
"replaces",
}
def test_minutes_for_exported() -> None:
assert fp.minutes_for(D1) == 0

173
focus_project/focus_project/tests/test_api.py Normal file
View File

@ -0,0 +1,173 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Registry, active/stands_down, create/end/list."""
from __future__ import annotations
from datetime import date, timedelta
import json
from typing import TYPE_CHECKING
from unittest import mock
import pytest
import focus_project as fp
from focus_project._errors import RegistryError
if TYPE_CHECKING:
from pathlib import Path
D1 = date(2026, 11, 2)
D4 = date(2026, 11, 5)
def _make(
pid: str = "hr", start: date = D1, end: date = D4, **kw: object
) -> fp.Project:
args: dict[str, object] = {
"name": "HackerRank prep",
"replaces": ["leetcode", "reading", "tutor"],
"daily_minutes": 120,
"unlock_minutes": 30,
}
args.update(kw)
return fp.create_project(pid, start=start, end=end, **args) # type: ignore[arg-type]
def test_empty_registry() -> None:
assert fp.list_projects() == []
assert fp.active(D1) is None
assert fp.stands_down("leetcode", D1) is False
def test_create_roundtrip_and_file_shape(data: Path) -> None:
p = _make(subjects=["a"], site_patterns=["leetcode.com"])
assert fp.list_projects() == [p]
raw = json.loads((data / "projects.json").read_text())
row = raw["projects"][0]
assert row["id"] == "hr"
assert row["start"] == "2026-11-02"
assert row["replaces"] == ["leetcode", "reading", "tutor"]
assert row["ended_on"] is None
assert row["created_ms"] > 0
assert row["subjects"] == ["a"]
assert not list(data.glob("*.tmp"))
def test_active_boundaries() -> None:
_make()
assert fp.active(D1 - timedelta(days=1)) is None
assert fp.active(D1).id == "hr" # type: ignore[union-attr]
assert fp.active(D4).id == "hr" # type: ignore[union-attr]
assert fp.active(D4 + timedelta(days=1)) is None
def test_stands_down_matrix() -> None:
_make()
assert fp.stands_down("leetcode", D1) is True
assert fp.stands_down("book", D1) is True
assert fp.stands_down("reading", D1) is True
assert fp.stands_down("tutor", D1) is True
assert fp.stands_down("workout", D1) is False
assert fp.stands_down("leetcode", D4 + timedelta(days=1)) is False
def test_book_alias_in_replaces() -> None:
_make(replaces=["book"])
assert fp.list_projects()[0].replaces == ("reading",)
assert fp.stands_down("book", D1) is True
assert fp.stands_down("leetcode", D1) is False
def test_overlap_rejected_and_adjacent_ok() -> None:
_make()
with pytest.raises(ValueError, match="overlap"):
_make("other", start=D4, end=D4 + timedelta(days=2))
_make("next", start=D4 + timedelta(days=1), end=D4 + timedelta(days=2))
assert len(fp.list_projects()) == 2
def test_end_early_frees_later_days() -> None:
_make()
ended = fp.end_project("hr", D1 + timedelta(days=1))
assert ended.ended_on == D1 + timedelta(days=1)
assert fp.active(D1 + timedelta(days=1)) is not None
assert fp.active(D1 + timedelta(days=2)) is None
assert fp.stands_down("leetcode", D1 + timedelta(days=2)) is False
_make("again", start=D1 + timedelta(days=2), end=D4)
def test_end_project_errors() -> None:
_make()
with pytest.raises(KeyError):
fp.end_project("nope", D1)
with pytest.raises(ValueError, match="before"):
fp.end_project("hr", D1 - timedelta(days=1))
fp.end_project("hr", D4 + timedelta(days=9)) # clamps to the planned end
assert fp.list_projects()[0].ended_on == D4
with pytest.raises(ValueError, match="already"):
fp.end_project("hr", D1)
@pytest.mark.parametrize(
("kw", "match"),
[
({"pid": "Bad Id"}, "id"),
({"end": D1 - timedelta(days=1)}, "end"),
({"replaces": ["workout"]}, "gate"),
({"replaces": []}, "replace"),
({"daily_minutes": 0}, "daily_minutes"),
({"unlock_minutes": 121}, "unlock_minutes"),
({"unlock_minutes": -1}, "unlock_minutes"),
({"name": " "}, "name"),
],
)
def test_create_validation(kw: dict[str, object], match: str) -> None:
with pytest.raises(ValueError, match=match):
_make(**kw) # type: ignore[arg-type]
assert fp.list_projects() == []
def test_duplicate_id_rejected() -> None:
_make()
with pytest.raises(ValueError, match="exists"):
_make(start=D4 + timedelta(days=5), end=D4 + timedelta(days=6))
def test_corrupt_registry_raises_but_stands_down_fails_closed(data: Path) -> None:
data.mkdir()
(data / "projects.json").write_text("{not json")
with pytest.raises(RegistryError):
fp.list_projects()
assert fp.stands_down("leetcode", D1) is False
@pytest.mark.parametrize(
"body", ['{"projects": [{"id": "x"}]}', "[]", '{"projects": 3}']
)
def test_bad_shapes(data: Path, body: str) -> None:
data.mkdir()
(data / "projects.json").write_text(body)
with pytest.raises(RegistryError):
fp.list_projects()
def test_overlapping_file_is_corrupt_for_active(data: Path) -> None:
_make()
raw = json.loads((data / "projects.json").read_text())
raw["projects"].append({**raw["projects"][0], "id": "dup"})
(data / "projects.json").write_text(json.dumps(raw))
with pytest.raises(RegistryError, match="overlap"):
fp.active(D1)
assert fp.stands_down("leetcode", D1) is False
def test_failed_write_keeps_old_file(data: Path) -> None:
_make()
before = (data / "projects.json").read_text()
with (
mock.patch("focus_project._store.os.replace", side_effect=OSError("boom")),
pytest.raises(OSError, match="boom"),
):
_make("two", start=D4 + timedelta(days=3), end=D4 + timedelta(days=4))
assert (data / "projects.json").read_text() == before
assert not list(data.glob("*.tmp"))

12
focus_project/install.sh Executable file
View File

@ -0,0 +1,12 @@
#!/bin/bash
# Editable user install so /usr/bin/python3 (the guards' user units) can
# `import focus_project`. Stdlib only; idempotent.
set -euo pipefail
REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
readonly REPO_DIR
readonly SYSTEM_PYTHON="/usr/bin/python3"
"$SYSTEM_PYTHON" -m pip install --user --no-deps --break-system-packages -e "$REPO_DIR"
cd /
"$SYSTEM_PYTHON" -c 'import focus_project; print(focus_project.__file__)'

171
focus_project/pyproject.toml Normal file
View File

@ -0,0 +1,171 @@
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[project]
name = "focus_project"
version = "0.1.0"
description = "Time-boxed focus projects: replace chosen gates, pay a daily-minutes gate"
requires-python = ">=3.14"
# Stdlib only on purpose: guards import it from the system python.
dependencies = []
[tool.setuptools.packages.find]
include = ["focus_project*"]
[tool.setuptools.package-data]
focus_project = ["py.typed"]
[tool.ruff]
target-version = "py314"
include = ["*.py", "**/*.py"]
exclude = [".git", ".venv", "__pycache__", "build", "dist", ".eggs"]
[tool.ruff.lint]
select = ["ALL"]
ignore = [
"D203",
"D213",
"COM812",
"ISC001",
"S603",
]
fixable = ["ALL"]
unfixable = []
[tool.ruff.lint.per-file-ignores]
"**/tests/**/*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
"**/test_*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
[tool.ruff.lint.pydocstyle]
convention = "google"
[tool.ruff.lint.isort]
force-single-line = false
force-sort-within-sections = true
known-first-party = ["focus_project"]
[tool.ruff.lint.flake8-quotes]
docstring-quotes = "double"
inline-quotes = "double"
[tool.ruff.lint.flake8-tidy-imports]
ban-relative-imports = "all"
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
skip-magic-trailing-comma = false
line-ending = "auto"
docstring-code-format = true
[tool.mypy]
python_version = "3.14"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
disallow_any_unimported = true
disallow_any_explicit = false
disallow_any_generics = true
disallow_subclassing_any = true
strict_equality = true
extra_checks = true
ignore_missing_imports = true
show_error_codes = true
color_output = true
exclude = [".venv/"]
[tool.pylint.main]
# The repo root on the path, so pylint resolves the package's own imports the
# same way pytest does instead of reporting import-error for them.
init-hook = "import sys; sys.path.insert(0, 'focus_project')"
analyse-fallback-blocks = true
persistent = true
jobs = 0
py-version = "3.14"
ignore = [".venv", "__pycache__"]
# Test suites intentionally use patterns (protected-access, fixture-arg
# shadowing) that don't apply to source code; ruff's per-file-ignores carry
# the equivalent exemptions, so tests stay linted -- by ruff. This is the same
# source-only pylint scope wake-alarm / diet-guard / testsAndMisc use. It has
# to be `ignore-paths`, not `ignore = ["tests"]`: `ignore` only matches the
# basename of what pylint is *given*, and pre-commit hands it explicit file
# paths, so a directory name in `ignore` never fires under the hook.
ignore-paths = ["^(.*/)?tests/.*$"]
ignore-patterns = [".*\\.pyi$"]
[tool.pylint.messages_control]
enable = "all"
disable = []
[tool.pylint.design]
min-public-methods = 0
max-module-lines = 1000
max-attributes = 10
[tool.pylint.typecheck]
generated-members = [
".*\\.assert_called_once_with",
".*\\.assert_called_once",
".*\\.assert_called",
".*\\.assert_not_called",
".*\\.assert_any_call",
".*\\.call_args",
".*\\.call_args_list",
".*\\.call_count",
]
[tool.bandit]
exclude_dirs = ["tests", ".venv"]
[tool.pytest.ini_options]
testpaths = ["focus_project/tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"-v",
"--strict-markers",
"--strict-config",
"-ra",
"--cov=focus_project",
"--cov-branch",
"--cov-report=term-missing",
"--cov-report=lcov",
]
filterwarnings = [
"error",
"ignore::DeprecationWarning",
"default::pytest.PytestUnraisableExceptionWarning",
]
[tool.coverage.run]
source = ["focus_project"]
branch = true
omit = [
"*/__pycache__/*",
"*/tests/*",
"*/.venv/*",
]
[tool.coverage.report]
fail_under = 100
show_missing = true
skip_covered = false
exclude_lines = [
"pragma: no cover",
"raise NotImplementedError",
"raise AssertionError",
"if TYPE_CHECKING:",
'if __name__ == "__main__":',
]
partial_branches = ["pragma: no branch"]

10
focus_project/requirements.txt Normal file
View File

@ -0,0 +1,10 @@
# Exact pins, enforced by the shared dependency-freshness gate: a `>=`
# floor is satisfied by anything newer, so it tells a checker nothing
# about whether this repo is actually current.
bandit[toml]==1.9.4
mypy==2.4.1
pre-commit==4.7.0
pylint==4.1.3
pytest==9.2.0
pytest-cov==7.1.0
ruff==0.17.3