kuhyx/screen-locker
1
0
Fork 0
mirror of https://github.com/kuhyx/screen-locker.git synced 2026-11-10 18:07:22 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Fix pylint findings in source and scripts

Browse Source 
Real fixes: split over-long functions (lock-decision chain, manual
sync, validation, phone verification, status), hoist a dead lazy
import, context-manage Xvfb and temp dirs, add encodings, share the
silent-failure argv loop.

lock_grab_handoff read lock._recovery, which gatelock's LockWindow
does not have (its public property is recovery); a contract test now
checks every attribute the handoff reads against the real class.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWYdwRAtMg7ki6TKxuU
This commit is contained in:
Krzysztof Rudnicki 2026-11-01 20:56:44 +01:00
parent 21a40ba9f5
commit f3875302d7
45 changed files with 583 additions and 449 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
screen_locker/_adb_transport.py
View File

@ -11,7 +11,7 @@ Deciding whether what came back counts as a workout stays in
from __future__ import annotations
from concurrent.futures import ( # pylint: disable=no-name-in-module
from concurrent.futures import (
ThreadPoolExecutor,
as_completed,
)

6
screen_locker/_armed_state.py
View File

@ -174,7 +174,7 @@ def locker_unit_active() -> bool | None:
if not systemctl_available():
return None
code, out = _run(["systemctl", "--user", "is-active", LOCKER_UNIT])
if code != 0 and not out.strip():
if code and not out.strip():
return None
return out.strip() == "active"
@ -186,7 +186,7 @@ def _login_target_active() -> bool:
True when ``graphical-session.target`` is active.
"""
code, out = _run(["systemctl", "--user", "is-active", _LOGIN_TARGET])
return code == 0 and out.strip() == "active"
return not code and out.strip() == "active"
def collect_states() -> list[TimerState]:
@ -205,7 +205,7 @@ def collect_states() -> list[TimerState]:
states.append(
TimerState(
name=name,
enabled=code == 0 and raw == "enabled",
enabled=not code and raw == "enabled",
enabled_raw=raw,
# For a timer this is "systemd has a next trigger for it";
# for the login-run service it is "the target it is wanted by

290
screen_locker/_compliance_state.py
View File

@ -18,6 +18,7 @@ surfaces the pending opportunity separately via ``AutoUpgradeOpportunity``.
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING
@ -59,6 +60,135 @@ __all__ = [
]
@dataclass(frozen=True)
class _Step:
"""One predicate of the lock-decision chain, in evaluation order."""
name: str
reason_true: str
reason_false: str
terminal_reason: str | None = None
note_when_false: bool = False
# Order mirrors ``_check_non_verify_exits`` / ``_check_today_state_exits``.
# ``early_bird_pending_expired`` is recorded but never terminal on its own: an
# expired pending marker still falls through toward a full lock, with the
# upgrade opportunity surfaced separately (see module docstring).
_CHAIN: tuple[_Step, ...] = (
_Step(
"scheduled_skip",
"Today is in scheduled_skips.json (unconditional manual skip).",
"Today is not a manually scheduled skip day.",
"Manually scheduled skip day — no lock, no workout required.",
),
_Step(
"early_bird_pending_expired",
"Logged in during the early-bird window and it has since closed "
"with no workout confirmed yet.",
"No expired early-bird pending marker.",
),
_Step(
"early_bird_window_open",
"Logged in during the early-bird window earlier today; still "
"waiting to see a workout before the window closes.",
"Not pending inside an open early-bird window.",
"Early-bird window still open — lock skipped while waiting for a workout.",
),
_Step(
"sick_day",
"Today is a sick day in the shared sickdays pool.",
"Today is not a sick day.",
"Sick day today — lock skipped.",
),
_Step(
"already_logged",
"A validly-signed workout is already logged for today.",
"No workout logged for today yet.",
"Workout already logged today — lock skipped.",
note_when_false=True,
),
_Step(
"wake_alarm_skip",
"The morning session earned a workout skip for now.",
"No morning-session skip in force.",
"Morning session earned a skip — lock skipped.",
),
_Step(
"relaxed_day_already_skipped",
"Relaxed day already dismissed via Skip — No Penalty today.",
"No same-day relaxed-day dismissal on record.",
"Relaxed day already skipped today — lock skipped, prompt not shown again.",
),
_Step(
"relaxed_day",
"Today is a relaxed day (Tue/Wed/Thu).",
"Today is not a relaxed day.",
"Relaxed day (Tue/Wed/Thu) — workout optional, lock skipped.",
),
_Step(
"weekly_minimum_met",
"Weekly workout minimum already reached.",
"Weekly workout minimum not yet reached.",
"Weekly workout minimum already met — lock skipped.",
),
_Step(
"evening_lockdown_today",
"Tonight's night lockdown already fired (>= 12:00 today).",
"No evening night lockdown yet today.",
"Night lockdown already happened tonight — the workout lock does not re-arm.",
),
)
def _chain_flags(
*,
log_file: Path,
weekly_minimum_met: bool,
relaxed_day: bool,
wake_skip: bool,
evening_lockdown: bool,
local_dt: datetime,
pending: bool,
) -> dict[str, bool]:
"""Evaluate every predicate once; the keys are the ``_CHAIN`` step names."""
today_str = local_day(local_dt)
# The early-bird window IS the wake-alarm carrot (EarlyBirdMixin).
return {
"scheduled_skip": is_scheduled_skip_today(local_dt.date()),
"early_bird_pending_expired": pending and not wake_skip,
"early_bird_window_open": pending and wake_skip,
"sick_day": is_sick_day_today(local_dt.date()),
"already_logged": has_logged_today(log_file, today=today_str),
"wake_alarm_skip": wake_skip,
"relaxed_day_already_skipped": relaxed_day
and is_relaxed_day_skipped_today(log_file, today=today_str),
"relaxed_day": relaxed_day,
"weekly_minimum_met": weekly_minimum_met,
"evening_lockdown_today": evening_lockdown,
}
def _walk_chain(
fired: dict[str, bool], context: StageContext, degraded_note: str
) -> LockExplanation | None:
"""Record each step in ``context.trace``; return the first terminal one."""
for step in _CHAIN:
hit = fired[step.name]
reason = step.reason_true if hit else step.reason_false
if step.note_when_false and not hit:
reason += degraded_note
context.trace.append(PredicateResult(step.name, hit, reason))
if hit and step.terminal_reason is not None:
return _stage(
fired=False,
stage=step.name,
reason=step.terminal_reason,
context=context,
)
return None
def explain_lock_decision(
*,
log_file: Path,
@ -77,155 +207,25 @@ def explain_lock_decision(
reported via ``heat_skip_evaluated=False``, not guessed.
"""
instant = now if now is not None else datetime.now(tz=UTC)
today_str = local_day(instant)
local_dt = instant.astimezone()
scheduled = is_scheduled_skip_today(local_dt.date())
pending = is_early_bird_pending(early_bird_pending_file, today=today_str)
# The early-bird window IS the wake-alarm carrot (EarlyBirdMixin).
window_open = wake_skip
expired = pending and not window_open
early_bird_open_and_pending = pending and window_open
sick_today = is_sick_day_today(local_dt.date())
logged = has_logged_today(log_file, today=today_str)
relaxed_day_already_skipped = relaxed_day and is_relaxed_day_skipped_today(
log_file, today=today_str
pending = is_early_bird_pending(early_bird_pending_file, today=local_day(instant))
fired = _chain_flags(
log_file=log_file,
weekly_minimum_met=weekly_minimum_met,
relaxed_day=relaxed_day,
wake_skip=wake_skip,
evening_lockdown=evening_lockdown,
local_dt=instant.astimezone(),
pending=pending,
)
auto_upgrade = describe_auto_upgrade_opportunity(
early_bird_pending=pending,
early_bird_window_open=window_open,
early_bird_window_open=wake_skip,
)
trace: list[PredicateResult] = []
stage_context = StageContext(trace, auto_upgrade, bool(degraded_sources))
def _check(
name: str,
*,
fired: bool,
reason_true: str,
reason_false: str,
terminal_reason: str | None = None,
) -> LockExplanation | None:
"""Record one trace step; return an explanation if it terminates the chain."""
reason = reason_true if fired else reason_false
trace.append(PredicateResult(name, fired, reason))
if fired and terminal_reason is not None:
return _stage(
fired=False,
stage=name,
reason=terminal_reason,
context=stage_context,
)
return None
result = _check(
"scheduled_skip",
fired=scheduled,
reason_true="Today is in scheduled_skips.json (unconditional manual skip).",
reason_false="Today is not a manually scheduled skip day.",
terminal_reason="Manually scheduled skip day — no lock, no workout required.",
)
if result is not None:
return result
# Recorded but never terminal on its own: an expired pending marker still
# falls through toward a full lock, with the upgrade opportunity surfaced
# separately via `auto_upgrade` (see module docstring).
_check(
"early_bird_pending_expired",
fired=expired,
reason_true="Logged in during the early-bird window and it has since closed "
"with no workout confirmed yet.",
reason_false="No expired early-bird pending marker.",
)
result = _check(
"early_bird_window_open",
fired=early_bird_open_and_pending,
reason_true="Logged in during the early-bird window earlier today; still "
"waiting to see a workout before the window closes.",
reason_false="Not pending inside an open early-bird window.",
terminal_reason=(
"Early-bird window still open — lock skipped while waiting for a workout."
),
)
if result is not None:
return result
result = _check(
"sick_day",
fired=sick_today,
reason_true="Today is a sick day in the shared sickdays pool.",
reason_false="Today is not a sick day.",
terminal_reason="Sick day today — lock skipped.",
)
if result is not None:
return result
context = StageContext([], auto_upgrade, bool(degraded_sources))
degraded_note = describe_degraded_sources(degraded_sources)
result = _check(
"already_logged",
fired=logged,
reason_true="A validly-signed workout is already logged for today.",
reason_false="No workout logged for today yet." + degraded_note,
terminal_reason="Workout already logged today — lock skipped.",
)
if result is not None:
return result
result = _check(
"wake_alarm_skip",
fired=wake_skip,
reason_true="The morning session earned a workout skip for now.",
reason_false="No morning-session skip in force.",
terminal_reason="Morning session earned a skip — lock skipped.",
)
if result is not None:
return result
result = _check(
"relaxed_day_already_skipped",
fired=relaxed_day_already_skipped,
reason_true="Relaxed day already dismissed via Skip — No Penalty today.",
reason_false="No same-day relaxed-day dismissal on record.",
terminal_reason="Relaxed day already skipped today — lock skipped, "
"prompt not shown again.",
)
if result is not None:
return result
result = _check(
"relaxed_day",
fired=relaxed_day,
reason_true="Today is a relaxed day (Tue/Wed/Thu).",
reason_false="Today is not a relaxed day.",
terminal_reason="Relaxed day (Tue/Wed/Thu) — workout optional, lock skipped.",
)
if result is not None:
return result
result = _check(
"weekly_minimum_met",
fired=weekly_minimum_met,
reason_true="Weekly workout minimum already reached.",
reason_false="Weekly workout minimum not yet reached.",
terminal_reason="Weekly workout minimum already met — lock skipped.",
)
if result is not None:
return result
result = _check(
"evening_lockdown_today",
fired=evening_lockdown,
reason_true="Tonight's night lockdown already fired (>= 12:00 today).",
reason_false="No evening night lockdown yet today.",
terminal_reason="Night lockdown already happened tonight — the workout "
"lock does not re-arm.",
)
if result is not None:
return result
terminal = _walk_chain(fired, context, degraded_note)
if terminal is not None:
return terminal
# Named for the conclusion; "..._pending_heat_check" read as "stuck".
return _stage(
@ -237,5 +237,5 @@ def explain_lock_decision(
"temperature is checked, and a run may or may not be happening "
"right now (see Health)." + degraded_note
),
context=stage_context,
context=context,
)

2
screen_locker/_credit_notify.py
View File

@ -103,7 +103,7 @@ def send_notification(title: str, body: str) -> bool:
except (OSError, subprocess.TimeoutExpired) as exc:
_logger.warning("Desktop notification failed (%s): %s", exc, body)
return False
if result.returncode != 0:
if result.returncode:
_logger.warning(
"notify-send exited %d (%s) — no desktop notification: %s",
result.returncode,

2
screen_locker/_credit_slot.py
View File

@ -57,7 +57,7 @@ def credit_slot(
if with_new == before:
# No new slot: either it earns nothing at all, or its slot is taken.
return CreditSlot.OCCUPIED if _counts_alone(new_entry) else CreditSlot.NONE
return CreditSlot.FIRST if before == 0 else CreditSlot.FURTHER
return CreditSlot.FURTHER if before else CreditSlot.FIRST
def _counts_alone(entry: Mapping[str, Any]) -> bool:

7
screen_locker/_extra_benefits.py
View File

@ -18,7 +18,7 @@ State is persisted in ``extra_benefits_state.json`` next to this file.
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from datetime import UTC, date, datetime, timedelta
import json
import logging
from typing import TYPE_CHECKING, Any
@ -62,7 +62,7 @@ def _save_state(state_file: Path, state: dict[str, Any]) -> None:
_logger.warning("Failed to save extra benefits state: %s", exc)
def _current_iso_week(now: datetime) -> str:
def _current_iso_week(now: date) -> str:
"""Return *now*'s ISO week as ``YYYY-Www``."""
year, week, _ = now.isocalendar()
return f"{year}-W{week:02d}"
@ -103,8 +103,7 @@ def process_week_transition(log_file: Path, state_file: Path) -> list[str]:
eb_weeks: list[str] = list(state.get("extended_early_bird_iso_weeks", []))
rewards: list[str] = []
prev_year, prev_week, _ = sunday_prev_week.isocalendar()
prev_week_str = f"{prev_year}-W{prev_week:02d}"
prev_week_str = _current_iso_week(sunday_prev_week)
if prev_week_count >= _BONUS_THRESHOLD:
streak += 1

2
screen_locker/_http_workout_fetch.py
View File

@ -10,7 +10,7 @@ app serves the same JSON over ``WORKOUT_HTTP_PORT``.
from __future__ import annotations
from concurrent.futures import ( # pylint: disable=no-name-in-module
from concurrent.futures import (
ThreadPoolExecutor,
as_completed,
)

6
screen_locker/_late_credit.py
View File

@ -54,11 +54,13 @@ class LateCreditMixin:
if TYPE_CHECKING:
def _read_shutdown_config(self) -> tuple[int, int, int] | None: ...
def _read_shutdown_config(self) -> tuple[int, int, int] | None:
raise NotImplementedError
def _write_shutdown_config(
self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = ...
) -> bool: ...
) -> bool:
raise NotImplementedError
def _apply_sick_floor(self) -> bool: ...

120
screen_locker/_manual_sync.py
View File

@ -45,6 +45,9 @@ if TYPE_CHECKING:
from collections.abc import Callable, Iterable, Mapping
from pathlib import Path
from screen_locker._log_mixin import RecordResult
from screen_locker._manual_workout import ManualWorkoutDraft
OnIngestedCallback = Callable[[dict, "list[dict]"], None]
__all__ = ["ingest_manual_records", "is_empty_stub", "reconstruct_draft"]
@ -113,6 +116,76 @@ def _due_date(
return date
def _new_valid_draft(
log_file: Path,
record_id: str,
payload: Mapping[str, object],
reference: str,
) -> tuple[str, ManualWorkoutDraft] | None:
"""The due date and valid draft of a not-yet-ingested record, else ``None``."""
date = _due_date(record_id, payload, reference)
if date is None:
return None
if _already_ingested(load_workout_log(log_file), record_id):
return None
draft = _draft_or_report(record_id, payload)
if draft is None:
return None
error = validate_manual_workout(draft)
if error is not None:
_logger.warning("Manual record %s invalid: %s", record_id, error)
return None
return date, draft
def _ingest_one(
log_file: Path,
record_id: str,
payload: Mapping[str, object],
reference: str,
pending: Mapping[str, str],
) -> tuple[dict[str, object], RecordResult] | None:
"""Validate, budget-check and append one record; ``None`` when skipped."""
ready = _new_valid_draft(log_file, record_id, payload, reference)
if ready is None:
return None
date, draft = ready
over_budget = budget_error_for_date(
log_file,
date,
[day for rid, day in pending.items() if rid != record_id],
)
if over_budget is not None:
_logger.info(
"Manual-workout budget exhausted — %s not ingested (%s)",
record_id,
over_budget,
)
return None
entry = build_entry(draft)
entry[_SYNC_ID_FIELD] = record_id
# The wire id IS the workout's id: the device that logged it minted
# ``manual:<date>T<start>`` once, and the publisher keys on the same
# id. Without this the chokepoint re-derived one from ``date`` +
# start, and when the two dates disagreed (a record whose day key was
# later corrected to the local day) the same workout landed twice
# under two ids -- and then consumed two manual-budget slots.
entry["workout_id"] = record_id
result = write_signed_entry(log_file, date, entry)
if not result.appended:
# This is dedup working, not credit being lost: the workout is
# already in the log under this same workout_id. It was logged at
# warning and re-reported on every 15-minute cycle, which read like
# a recurring fault and buried the lines that did need attention.
_logger.info(
"Manual record %s is already logged under the same workout_id "
"— skipped as a duplicate (its credit is already counted)",
record_id,
)
return None
return entry, result
def ingest_manual_records(
log_file: Path,
records: Iterable[tuple[str, Mapping[str, object]]],
@ -157,51 +230,10 @@ def ingest_manual_records(
pending = pending_future_dates(log_file, batch, today=reference)
ingested: list[str] = []
for record_id, payload in batch:
date = _due_date(record_id, payload, reference)
if date is None:
continue
if _already_ingested(load_workout_log(log_file), record_id):
continue
draft = _draft_or_report(record_id, payload)
if draft is None:
continue
error = validate_manual_workout(draft)
if error is not None:
_logger.warning("Manual record %s invalid: %s", record_id, error)
continue
over_budget = budget_error_for_date(
log_file,
date,
[day for rid, day in pending.items() if rid != record_id],
)
if over_budget is not None:
_logger.info(
"Manual-workout budget exhausted — %s not ingested (%s)",
record_id,
over_budget,
)
continue
entry = build_entry(draft)
entry[_SYNC_ID_FIELD] = record_id
# The wire id IS the workout's id: the device that logged it minted
# ``manual:<date>T<start>`` once, and the publisher keys on the same
# id. Without this the chokepoint re-derived one from ``date`` +
# start, and when the two dates disagreed (a record whose day key was
# later corrected to the local day) the same workout landed twice
# under two ids -- and then consumed two manual-budget slots.
entry["workout_id"] = record_id
result = write_signed_entry(log_file, date, entry)
if not result.appended:
# This is dedup working, not credit being lost: the workout is
# already in the log under this same workout_id. It was logged at
# warning and re-reported on every 15-minute cycle, which read like
# a recurring fault and buried the lines that did need attention.
_logger.info(
"Manual record %s is already logged under the same workout_id "
"— skipped as a duplicate (its credit is already counted)",
record_id,
)
outcome = _ingest_one(log_file, record_id, payload, reference, pending)
if outcome is None:
continue
entry, result = outcome
ingested.append(record_id)
if on_ingested is not None:
on_ingested(entry, result.prior_entries)

2
screen_locker/_manual_sync_draft.py
View File

@ -49,7 +49,7 @@ def is_empty_stub(payload: Mapping[str, object]) -> bool:
produced an unbounded stream of identical warnings (four every 15 minutes),
which is how the genuinely actionable lines got lost in the noise.
"""
return not (_SUBSTANTIVE_FIELDS & set(payload))
return not _SUBSTANTIVE_FIELDS & set(payload)
def _coerce_int(value: object) -> int:

36
screen_locker/_manual_workout.py
View File

@ -20,12 +20,15 @@ import logging
from screen_locker._constants import (
MANUAL_WORKOUT_DESCRIPTION_MIN_CHARS,
MANUAL_WORKOUT_REFLECTION_MIN_CHARS,
MANUAL_WORKOUT_RPE_MAX,
MANUAL_WORKOUT_RPE_MIN,
MIN_WORKOUT_DURATION_MINUTES,
WORKOUT_DURATION_ACCEPT_MINUTES,
)
from screen_locker._manual_workout_checks import (
validate_reflections,
validate_required,
)
from screen_locker._manual_workout_types import (
MANUAL_WORKOUT_TYPE,
SPORT_CHOICES,
@ -82,19 +85,9 @@ def _duration_minutes(draft: ManualWorkoutDraft) -> float | None:
def validate_manual_workout(draft: ManualWorkoutDraft) -> str | None:
"""Return an error message if the draft is invalid, else None."""
if draft.sport not in SPORT_CHOICES:
return "Please choose a sport"
required = {
"Start time": draft.start_time,
"End time": draft.end_time,
"Location name": draft.location_name,
"Transport method": draft.transport_method,
"Cost": draft.cost,
}
for label, value in required.items():
if not value.strip():
return f"{label} is required"
error = validate_required(draft)
if error is not None:
return error
duration = _duration_minutes(draft)
if duration is None:
@ -119,18 +112,7 @@ def validate_manual_workout(draft: ManualWorkoutDraft) -> str | None:
if error is not None:
return error
for label, value in (
("What went well", draft.went_well),
("What to improve", draft.to_improve),
("Overall feeling", draft.overall_feeling),
):
if len(value.strip()) < MANUAL_WORKOUT_REFLECTION_MIN_CHARS:
return (
f"{label} must be at least {MANUAL_WORKOUT_REFLECTION_MIN_CHARS} "
f"characters (currently {len(value.strip())})"
)
return None
return validate_reflections(draft)
def _validate_table_tennis(draft: ManualWorkoutDraft) -> str | None:
@ -143,7 +125,7 @@ def _validate_table_tennis(draft: ManualWorkoutDraft) -> str | None:
):
if value < 0:
return f"{label} cannot be negative"
if draft.matches_won + draft.matches_lost == 0:
if not draft.matches_won + draft.matches_lost:
return "Enter at least one match played (won + lost)"
if not draft.racket.strip():
return "Racket is required"

44
screen_locker/_manual_workout_checks.py Normal file
View File

@ -0,0 +1,44 @@
"""Field-level checks shared by ``validate_manual_workout``.
Split out of :mod:`screen_locker._manual_workout` to keep every file under the
250-line cap.
"""
from __future__ import annotations
from screen_locker._constants import MANUAL_WORKOUT_REFLECTION_MIN_CHARS
from screen_locker._manual_workout_types import SPORT_CHOICES, ManualWorkoutDraft
def validate_required(draft: ManualWorkoutDraft) -> str | None:
"""Return an error if the sport is unchosen or a required field is blank."""
if draft.sport not in SPORT_CHOICES:
return "Please choose a sport"
required = {
"Start time": draft.start_time,
"End time": draft.end_time,
"Location name": draft.location_name,
"Transport method": draft.transport_method,
"Cost": draft.cost,
}
for label, value in required.items():
if not value.strip():
return f"{label} is required"
return None
def validate_reflections(draft: ManualWorkoutDraft) -> str | None:
"""Return an error if a reflection field is shorter than the minimum."""
for label, value in (
("What went well", draft.went_well),
("What to improve", draft.to_improve),
("Overall feeling", draft.overall_feeling),
):
if len(value.strip()) < MANUAL_WORKOUT_REFLECTION_MIN_CHARS:
return (
f"{label} must be at least {MANUAL_WORKOUT_REFLECTION_MIN_CHARS} "
f"characters (currently {len(value.strip())})"
)
return None

4
screen_locker/_manual_workout_types.py
View File

@ -103,8 +103,8 @@ def count_in_window(
dated after today are NOT in it. Whether a new entry fits is a different
question, answered per workout date by ``_manual_future``.
"""
today_str = today or _today_iso()
today_dt = _parse_iso(today_str)
day = today or _today_iso()
today_dt = _parse_iso(day)
if today_dt is None:
return 0
cutoff = today_dt - timedelta(days=days)

105
screen_locker/_phone_verification.py
View File

@ -28,6 +28,18 @@ _HTTP_OK = _http_client.OK
_logger = logging.getLogger(__name__)
def _in_current_week(date_str: str) -> bool:
"""True if *date_str* parses and falls in this ISO week up to today."""
today = datetime.now(tz=UTC).astimezone().date()
week_start = today - timedelta(days=today.weekday())
try:
entry_date = date.fromisoformat(date_str)
except ValueError:
_logger.warning("StrongLifts JSON has an unparsable date: %r", date_str)
return False
return week_start <= entry_date <= today
class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin):
"""Mixin providing phone-based workout verification via ADB."""
@ -119,16 +131,7 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin):
return 0
data = self._pull_workout_app_json()
date_str = data.get("date") if data else None
if not date_str:
return 0
today = datetime.now(tz=UTC).astimezone().date()
week_start = today - timedelta(days=today.weekday())
try:
entry_date = date.fromisoformat(date_str)
except ValueError:
_logger.warning("StrongLifts JSON has an unparsable date: %r", date_str)
return 0
if not week_start <= entry_date <= today:
if not date_str or not _in_current_week(date_str):
return 0
status, msg = self._validate_json_content(data)
if status != "verified":
@ -145,6 +148,33 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin):
# ── HTTP fallback (no ADB / developer options required) ───────────────────
def _pull_sync_result(self) -> tuple[tuple[str, str] | None, str | None]:
"""Firebase sync first: the validated result (if any) and sync error.
Firebase sync is the primary channel -- it works without the phone
being on the same network as the PC. Only a *verified* result
short-circuits ADB/HTTP: a stale/incomplete cloud entry (e.g. the
phone's last push predates today, or a later push failed offline)
must not preempt a fresher ADB pull, so anything less than
"verified" is kept by the caller as a fallback-of-last-resort
candidate. A sync error never blocks unlocking either; `sync_failed`
only surfaces if ADB/HTTP also come up empty.
"""
synced_data, sync_error = pull_synced_workout()
if synced_data is None:
if sync_error is not None:
_logger.info(
"Firebase sync unavailable (%s) — trying ADB/HTTP...", sync_error
)
return None, sync_error
sync_result = self._validate_json_data(synced_data)
if sync_result[0] != "verified":
_logger.info(
"Synced data not fully verified (%s) — trying ADB/HTTP...",
sync_result[0],
)
return sync_result, sync_error
def _verify_phone_workout(self) -> tuple[str, str]:
"""Verify today's workout: Firebase sync first, ADB/HTTP as fallback.
@ -156,28 +186,9 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin):
if not clock_ok:
return "clock_tampered", clock_msg
# Firebase sync is the primary channel — it works without the phone
# being on the same network as the PC. Only a *verified* sync result
# short-circuits ADB/HTTP: a stale/incomplete cloud entry (e.g. the
# phone's last push predates today, or a later push failed offline)
# must not preempt a fresher ADB pull, so anything less than
# "verified" is kept as a fallback-of-last-resort candidate instead
# of being returned immediately. A sync error never blocks unlocking
# either; `sync_failed` only surfaces if ADB/HTTP also come up empty.
synced_data, sync_error = pull_synced_workout()
sync_result: tuple[str, str] | None = None
if synced_data is not None:
sync_result = self._validate_json_data(synced_data)
if sync_result[0] == "verified":
return sync_result
_logger.info(
"Synced data not fully verified (%s) — trying ADB/HTTP...",
sync_result[0],
)
elif sync_error is not None:
_logger.info(
"Firebase sync unavailable (%s) — trying ADB/HTTP...", sync_error
)
sync_result, sync_error = self._pull_sync_result()
if sync_result is not None and sync_result[0] == "verified":
return sync_result
# Prefer ADB when a device is visible, but if the pull yields no usable
# JSON, fall through to the HTTP/WiFi scan — the app's in-memory HTTP
@ -194,13 +205,23 @@ class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin):
data = self._fetch_http_workout()
if data is not None:
return self._validate_json_data(data)
if sync_result is not None:
return sync_result
if sync_error is not None:
return "sync_failed", sync_error
if adb_connected:
return (
"not_verified",
"Workout app JSON not found. Complete a workout in the app first.",
)
return "no_phone", "Phone not reachable via ADB or HTTP on local network"
return _no_data_result(sync_result, sync_error, adb_connected=adb_connected)
def _no_data_result(
sync_result: tuple[str, str] | None,
sync_error: str | None,
*,
adb_connected: bool,
) -> tuple[str, str]:
"""The verdict when neither ADB nor HTTP produced a workout JSON."""
if sync_result is not None:
return sync_result
if sync_error is not None:
return "sync_failed", sync_error
if adb_connected:
return (
"not_verified",
"Workout app JSON not found. Complete a workout in the app first.",
)
return "no_phone", "Phone not reachable via ADB or HTTP on local network"

2
screen_locker/_poke_server.py
View File

@ -135,7 +135,7 @@ class PokeHandler(BaseHTTPRequestHandler):
started = time.perf_counter()
try:
status, reply = self._handle()
except Exception:
except Exception: # pylint: disable=broad-exception-caught # handler boundary: logged, answered 500
_logger.exception("Workout poke from %s crashed", self.client_address[0])
status = HTTPStatus.INTERNAL_SERVER_ERROR
reply = _refused(

2
screen_locker/_session_stream.py
View File

@ -123,7 +123,7 @@ class SessionStream:
except StreamError as exc:
_logger.warning("Firebase session stream failed: %s; %s", exc, _FALLBACK)
return False
except Exception:
except Exception: # pylint: disable=broad-exception-caught # daemon boundary: logged, loop reconnects
_logger.exception("Firebase session stream crashed; %s", _FALLBACK)
return False
return True

3
screen_locker/_shutdown.py
View File

@ -63,8 +63,7 @@ def read_shutdown_config(path: Path) -> tuple[int, int, int] | None:
else:
_logger.warning("Shutdown config missing required values")
return None
mon_wed, thu_sun, morning_end = values
return mon_wed, thu_sun, morning_end
return values[0], values[1], values[2]
class ShutdownMixin(SickFloorMixin, LateCreditMixin):

6
screen_locker/_sick_floor.py
View File

@ -48,11 +48,13 @@ class SickFloorMixin:
if TYPE_CHECKING:
def _read_shutdown_config(self) -> tuple[int, int, int] | None: ...
def _read_shutdown_config(self) -> tuple[int, int, int] | None:
raise NotImplementedError
def _write_shutdown_config(
self, mon_wed: int, thu_sun: int, morning_end: int, *, restore: bool = ...
) -> bool: ...
) -> bool:
raise NotImplementedError
def _apply_sick_floor(self) -> bool:
"""Raise tonight's shutdown to the sick target if today is sick.

108
screen_locker/_status.py
View File

@ -71,55 +71,29 @@ def _print_day_line(d: date, entries: list[dict], sick_days: frozenset[date]) ->
return any_counted
def run_status(locker: ScreenLocker) -> None:
"""Print weekly workout status, run RunnerUp scan, apply bonus, then exit."""
today = datetime.now(tz=UTC).astimezone().date()
monday = today - timedelta(days=today.weekday())
log_file: Path = locker.log_file
log_data = load_workout_log(log_file)
sick_days = workout_standdown_dates()
def _scan_runnerup(locker: ScreenLocker, log_file: Path, before_count: int) -> int:
"""Run the RunnerUp week scan, print what it filled, return the new count.
print("=== Weekly Workout Status ===")
# Per-day breakdown
for i in range(7):
d = monday + timedelta(days=i)
if d > today:
break
_print_day_line(d, log_data.get(d.isoformat(), []), sick_days)
print()
# The weekly total, per the anti-gaming rule (each verified workout counts
# individually, manual counts once/day) — NOT a per-day checkmark count,
# which would undercount a day holding two verified workouts.
before_count = count_weekly_workouts(log_file)
# RunnerUp scan. Each fill is credited by the shared ingestion callback
# (+2h first counted workout of its day, +1h each further one), the same
# rule the timer's sync pass and the daily base reset use.
Each fill is credited by the shared ingestion callback (+2h first counted
workout of its day, +1h each further one), the same rule the timer's sync
pass and the daily base reset use.
"""
n_filled = locker._scan_and_fill_week_runnerup(
log_file, on_ingested=locker._credit_ingested_workout
)
locker.workout_data = {}
if n_filled > 0:
print(f" Auto-filled {n_filled} workout(s) from RunnerUp exports.")
after_count = count_weekly_workouts(log_file)
_, thu_sun, _ = locker._read_shutdown_config() or (None, None, None)
if thu_sun is not None:
print(f" Shutdown tonight: {hhmm(thu_sun)}.")
else:
if n_filled <= 0:
print(" No new workouts found via RunnerUp scan.")
after_count = before_count
return before_count
print(f" Auto-filled {n_filled} workout(s) from RunnerUp exports.")
_, thu_sun, _ = locker._read_shutdown_config() or (None, None, None)
if thu_sun is not None:
print(f" Shutdown tonight: {hhmm(thu_sun)}.")
return count_weekly_workouts(log_file)
print()
# Extra benefits summary
streak = current_streak(EXTRA_BENEFITS_FILE)
eb_ext = has_extended_early_bird(EXTRA_BENEFITS_FILE)
eb_str = "Yes — until 09:00" if eb_ext else "No"
# Heat skips this month
def _heat_skips_this_month(log_data: dict[str, list[dict]]) -> str:
"""One-line summary of this month's heat skips: count and the latest."""
this_month = datetime.now(tz=UTC).astimezone().date().strftime("%Y-%m")
heat_entries = [
(d, entry)
@ -129,21 +103,17 @@ def run_status(locker: ScreenLocker) -> None:
if isinstance(entry, dict)
and entry.get("workout_data", {}).get("type") == "heat_skip"
]
if heat_entries:
last_date, last_e = max(heat_entries, key=lambda de: de[0])
last_temp = last_e.get("workout_data", {}).get("temperature_celsius", "?")
heat_str = f"{len(heat_entries)} (last: {last_date}, {last_temp}°C)"
else:
heat_str = "0"
if not heat_entries:
return "0"
last_date, last_e = max(heat_entries, key=lambda de: de[0])
last_temp = last_e.get("workout_data", {}).get("temperature_celsius", "?")
return f"{len(heat_entries)} (last: {last_date}, {last_temp}°C)"
print(f" Streak (5+ wks) : {streak}")
print(f" Early-bird extended : {eb_str}")
print(f" Heat skips (month) : {heat_str}")
print()
def _print_minimum(after_count: int) -> None:
"""Print how far the weekly count is from (or past) the minimum."""
remaining = max(0, WEEKLY_WORKOUT_MINIMUM - after_count)
extra = max(0, after_count - WEEKLY_WORKOUT_MINIMUM)
if remaining > 0:
print(
f" Need {remaining} more to reach the minimum ({WEEKLY_WORKOUT_MINIMUM})."
@ -156,13 +126,37 @@ def run_status(locker: ScreenLocker) -> None:
f" ({WEEKLY_WORKOUT_MINIMUM}/{WEEKLY_WORKOUT_MINIMUM})."
)
# Shutdown config
def run_status(locker: ScreenLocker) -> None:
"""Print weekly workout status, run RunnerUp scan, apply bonus, then exit."""
today = datetime.now(tz=UTC).astimezone().date()
monday = today - timedelta(days=today.weekday())
log_file: Path = locker.log_file
log_data = load_workout_log(log_file)
sick_days = workout_standdown_dates()
print("=== Weekly Workout Status ===")
for i in range(7):
d = monday + timedelta(days=i)
if d > today:
break
_print_day_line(d, log_data.get(d.isoformat(), []), sick_days)
print()
after_count = _scan_runnerup(locker, log_file, count_weekly_workouts(log_file))
print()
eb_ext = has_extended_early_bird(EXTRA_BENEFITS_FILE)
print(f" Streak (5+ wks) : {current_streak(EXTRA_BENEFITS_FILE)}")
print(f" Early-bird extended : {'Yes — until 09:00' if eb_ext else 'No'}")
print(f" Heat skips (month) : {_heat_skips_this_month(log_data)}")
print()
_print_minimum(after_count)
cfg = locker._read_shutdown_config()
if cfg:
_mw, _ts, _morning = cfg
print(f" Shutdown tonight : {hhmm(_mw)}")
print(f" Shutdown tonight : {hhmm(cfg[0])}")
# --status is a read-out, not an enforcement run; make that explicit so the
# journal never reads as though the locker evaluated and declined to lock.
record_no_decision("--status")
sys.exit(0)

2
screen_locker/_status_projection.py
View File

@ -89,7 +89,7 @@ def _tonight_minutes(snapshot: StatusSnapshot) -> int | None:
def format_summary_line(snapshot: StatusSnapshot) -> str:
"""Cheap one-liner for the i3blocks summary (e.g. before/after a click)."""
week = snapshot.week
mark = "✓" if week.remaining == 0 else "…"
mark = "…" if week.remaining else "✓"
tonight = _tonight_minutes(snapshot)
tonight_str = hhmm(tonight) if tonight is not None else "?"
sick = " · sick day" if snapshot.today.is_sick_day else ""

6
screen_locker/_status_view_verify.py
View File

@ -8,7 +8,7 @@ screen's unlock does.
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import date
from typing import TYPE_CHECKING
@ -19,6 +19,7 @@ from screen_locker._day import today_str
from screen_locker._earned import extra_minutes, span
from screen_locker._status_data import gather_status
from screen_locker._weekly_check import WEEKLY_WORKOUT_MINIMUM, count_weekly_workouts
from screen_locker.screen_lock import ScreenLocker
if TYPE_CHECKING:
from collections.abc import Callable
@ -26,7 +27,6 @@ if TYPE_CHECKING:
from pathlib import Path
from screen_locker._workout_credit import WorkoutCreditResult
from screen_locker.screen_lock import ScreenLocker
@dataclass(frozen=True)
@ -53,8 +53,6 @@ def _make_bare_verifier(log_file: Path) -> ScreenLocker:
``--status`` — just enough state for the verification mixins' methods
to run, no Tk lock UI, no ``__init__`` side effects.
"""
from screen_locker.screen_lock import ScreenLocker
verifier = object.__new__(ScreenLocker)
verifier.log_file = log_file
verifier.workout_data = {}

2
screen_locker/_stream_cache.py
View File

@ -128,7 +128,7 @@ class DeviceLogCache:
outcome = _poke_credit.credit_sessions(
[(rid, payload) for rid, payload, _hlc in pending]
)
except Exception:
except Exception: # pylint: disable=broad-exception-caught # one bad event must not drop the stream
# Not a stream problem: keep the connection (a reconnect would
# re-download the snapshot and fail the same way). Nothing is
# marked seen, so the next event retries.

2
screen_locker/_ui_flows.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module
from concurrent.futures import ThreadPoolExecutor
from typing import TYPE_CHECKING
from screen_locker import _manual_workout

2
screen_locker/_ui_flows_relaxed.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor # pylint: disable=no-name-in-module
from concurrent.futures import ThreadPoolExecutor
from screen_locker._weekly_check import (
RELAXED_DAY_SKIP_TYPE,

2
screen_locker/_web_server.py
View File

@ -89,7 +89,7 @@ def _decision_limit(query: str) -> int:
class _Handler(BaseHTTPRequestHandler):
"""Serve the status payloads and the static frontend bundle (read-only)."""
def log_message(self, fmt: str, *args: object) -> None:
def log_message(self, fmt: str, /, *args: object) -> None:
"""Route the default request log to ``logging`` at debug level.
Args:

2
screen_locker/_weekly_check.py
View File

@ -223,7 +223,7 @@ def count_weekly_workouts(
exc,
)
continue
if not (week_start <= entry_date <= today_date):
if entry_date < week_start or entry_date > today_date:
continue
count += count_day_credits(date_str, entries)
return count

6
screen_locker/_workout_app.py
View File

@ -49,9 +49,11 @@ if TYPE_CHECKING:
stdout: IO[str] | None
returncode: int | None
def terminate(self) -> None: ...
def terminate(self) -> None:
"""Ask the process to stop."""
def wait(self) -> int: ...
def wait(self) -> int:
"""Block until the process exits; return its exit code."""
class _Spawner(Protocol):
"""``subprocess.Popen``, narrowed to the one call shape used here."""

2
screen_locker/_workout_credit.py
View File

@ -183,7 +183,7 @@ class WorkoutCreditMixin:
"""
try:
_credit_notify.notify_workout_credit(before, after, self.log_file)
except Exception:
except Exception: # pylint: disable=broad-exception-caught # notification must never fail the credit
_logger.exception(
"Workout credit notification failed — the credit itself is "
"applied; only the desktop notification is missing"

2
screen_locker/_workout_handoff.py
View File

@ -41,7 +41,7 @@ def lock_grab_handoff(lock: LockWindow) -> GrabHandoff:
def release() -> None:
# Stop first: a watchdog tick between the release and the app's own
# grab would take the screen straight back off it.
lock._recovery.stop()
lock.recovery.stop()
# Claim.grab_held false BEFORE the release (fail closed), so
# leisure-watch never reads "held" while the workout app has input.
# reacquire() -> grab_input() sets it true once the grab lands.

6
screen_locker/_workout_session.py
View File

@ -38,9 +38,11 @@ if TYPE_CHECKING:
stdout: IO[str] | None
returncode: int | None
def poll(self) -> int | None: ...
def poll(self) -> int | None:
"""Return the exit code, or ``None`` while it still runs."""
def terminate(self) -> None: ...
def terminate(self) -> None:
"""Ask the process to stop."""
__all__ = ["WorkoutSession"]

2
screen_locker/poke_server.py
View File

@ -69,7 +69,7 @@ def _start_stream(server: PokeServer) -> None:
# Same lock as the HTTP handler: a poke and a streamed copy of one
# session must never be credited concurrently.
SessionStream(server.gate).start()
except Exception:
except Exception: # pylint: disable=broad-exception-caught # daemon boundary: logged, LAN poke must survive
_logger.exception(
"Could not start the Firebase session stream; off-LAN workouts will "
"only be credited by the 15-min workout-sync until the daemon restarts"

41
screen_locker/tests/test_workout_handoff.py
View File

@ -8,6 +8,9 @@ the user Ctrl+Alt+F2 out of the obligation the lock exists to enforce.
from __future__ import annotations
import tkinter as tk
from unittest.mock import MagicMock
from gatelock import LockWindow
from screen_locker._workout_handoff import lock_grab_handoff, start_workout
@ -28,7 +31,7 @@ class _FakeLock:
"""A lock window that records the order of grab operations."""
def __init__(self, *, release_raises: bool = False) -> None:
self._recovery = _FakeRecovery()
self.recovery = _FakeRecovery()
self.events: list[str] = []
self.closed = False
self._release_raises = release_raises
@ -49,7 +52,7 @@ class _FakeLock:
def grab_input(self) -> None:
self.events.append("grab_input")
self._recovery.running = True
self.recovery.running = True
def close(self) -> None:
self.closed = True
@ -62,16 +65,16 @@ class TestLockGrabHandoff:
"""A tick between release and the app's own grab would steal it back."""
lock = _FakeLock()
lock_grab_handoff(lock).release()
assert lock._recovery.events == ["stop"]
assert lock.recovery.events == ["stop"]
assert lock.events == ["held=False", "grab_release"]
assert not lock._recovery.running
assert not lock.recovery.running
def test_reacquire_restores_the_grab_and_the_watchdog(self) -> None:
handoff = lock_grab_handoff(lock := _FakeLock())
handoff.release()
handoff.reacquire()
assert lock.events == ["held=False", "grab_release", "grab_input"]
assert lock._recovery.running
assert lock.recovery.running
def test_never_closes_the_lock_window(self) -> None:
"""close() also restores VT switching -- that is the escape hatch."""
@ -116,3 +119,31 @@ class TestStartWorkout:
_workout_handoff, "launch_workout_app", lambda _h: _Result()
)
assert "checking whether it counted" in start_workout(_FakeLock())
class _Recorder:
"""A lock stand-in that records every attribute the handoff reads."""
def __init__(self) -> None:
self.names: set[str] = set()
def __getattr__(self, name: str) -> MagicMock:
self.names.add(name)
return MagicMock()
class TestAgainstTheRealGatelockApi:
"""The fakes above cannot notice a renamed or private gatelock attribute."""
def test_the_handoff_only_reads_attributes_a_real_lock_window_has(self) -> None:
"""``lock._recovery`` raised AttributeError live while fakes passed."""
spy = _Recorder()
handoff = lock_grab_handoff(spy)
handoff.release()
handoff.reacquire()
missing = {
name
for name in spy.names
if name != "root" and not hasattr(LockWindow, name)
}
assert not missing

13
scripts/_popup_form_check.py
View File

@ -130,7 +130,7 @@ def grab_holder(root: tk.Tk) -> str:
class FormHost(UIWidgetsMixin, ManualWorkoutDialogMixin):
"""The two hooks the mixin requires, over a real Tk root."""
def __init__(self, root: tk.Tk) -> None:
def __init__(self, root: tk.Tk, log_file: Path) -> None:
"""Set only the attributes the form actually reads."""
self.root = root
self.demo_mode = False
@ -139,8 +139,7 @@ class FormHost(UIWidgetsMixin, ManualWorkoutDialogMixin):
# week's manual workouts are used up, so pointing at the live log made
# this check pass or fail on how much the user had exercised that
# week rather than on whether the widgets survive a recovery tick.
self._log_dir = TemporaryDirectory()
self.log_file = Path(self._log_dir.name) / "log.json"
self.log_file = log_file
self.log_file.write_text("{}")
self._colors = LockConfig()
self.container = FrameGroup.single(root, bg=self._colors.palette.bg)
@ -178,9 +177,15 @@ def radios(widget: tk.Misc) -> list[tk.Radiobutton]:
def check_real_form_survives() -> bool:
"""Build the production form, pick "Other", hold it through the ticks."""
with TemporaryDirectory() as scratch:
return _check_form(Path(scratch) / "log.json")
def _check_form(log_file: Path) -> bool:
"""The body of :func:`check_real_form_survives`, over a throwaway log."""
root = tk.Tk()
root.geometry("1600x1200+0+0")
host = FormHost(root)
host = FormHost(root, log_file)
host.render()
root.update()

4
scripts/_preview_sim.py
View File

@ -33,8 +33,8 @@ def stamp(day: date, hhmm_text: str) -> float:
return moment.timestamp()
def _iso(stamp: float) -> str:
return datetime.fromtimestamp(stamp).astimezone().isoformat()
def _iso(moment: float) -> str:
return datetime.fromtimestamp(moment).astimezone().isoformat()
def write_walk(folder: Path, day: date, start: str, minutes: int) -> None:

40
scripts/_silent_cli.py Normal file
View File

@ -0,0 +1,40 @@
"""The argv loop shared by the Python and Dart silent-failure hooks.
Both hooks read every file named on the command line, report what their
``check_source`` finds, and fail closed on a file they cannot read. Only the
per-file check differs, so it is the parameter.
"""
from __future__ import annotations
import logging
from pathlib import Path
import sys
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable
_logger = logging.getLogger(__name__)
def run_hook(check_source: Callable[[Path, str], list[str]]) -> int:
"""Return 1 if ``check_source`` flags any file in ``sys.argv``, else 0."""
problems: list[str] = []
for filename in sys.argv[1:]:
path = Path(filename)
try:
source = path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
_logger.warning(
"Could not read %s (%s) — reporting it as a problem rather than "
"skipping it, since its handlers cannot be checked",
path,
exc,
)
problems.append(f"{path}: could not read ({exc})")
continue
problems.extend(check_source(path, source))
for problem in problems:
sys.stdout.write(f"{problem}\n")
return 1 if problems else 0

32
scripts/_xvfb_reexec.py
View File

@ -13,12 +13,17 @@ forever.
from __future__ import annotations
import contextlib
import logging
import os
from pathlib import Path
import shutil
import subprocess
import sys
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterator
_logger = logging.getLogger(__name__)
@ -63,8 +68,7 @@ def reexec_under_xvfb(
repo_root = Path(__file__).resolve().parent.parent
# Absolute argv, no shell.
target = ["-m", module] if module is not None else [str(script)]
server, display = _start_xvfb(xvfb)
try:
with _xvfb_display(xvfb) as display:
if display is None:
_logger.error("Xvfb exited before it was ready to accept clients")
return 1
@ -73,26 +77,28 @@ def reexec_under_xvfb(
return subprocess.run(
[sys.executable, *target], check=False, cwd=repo_root, env=env
).returncode
finally:
server.terminate()
server.wait(timeout=5)
def _start_xvfb(xvfb: str) -> tuple[subprocess.Popen[bytes], str | None]:
"""Start Xvfb on a free display; return it and ``:N`` once it is ready.
@contextlib.contextmanager
def _xvfb_display(xvfb: str) -> Iterator[str | None]:
"""Run Xvfb on a free display; yield ``:N`` once ready (``None`` if it died).
``-displayfd`` makes Xvfb pick a free display and write its number only
once it accepts clients. ``xvfb-run`` instead sleeps a fixed 3 s after
starting it, which was most of this pre-commit hook's runtime.
"""
read_fd, write_fd = os.pipe()
server = subprocess.Popen(
with subprocess.Popen(
[xvfb, "-displayfd", str(write_fd), "-nolisten", "tcp", *_SCREEN_ARG],
pass_fds=(write_fd,),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
os.close(write_fd)
with os.fdopen(read_fd) as ready:
number = ready.readline().strip()
return server, f":{number}" if number else None
) as server:
os.close(write_fd)
try:
with os.fdopen(read_fd) as ready:
number = ready.readline().strip()
yield f":{number}" if number else None
finally:
server.terminate()
server.wait(timeout=5)

2
scripts/check_requirements_sync.py
View File

@ -88,7 +88,7 @@ def _top_level_imports(module: Path) -> set[str]:
if isinstance(node, ast.Import):
for alias in node.names:
imported.add(alias.name.split(".", 1)[0])
elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
elif isinstance(node, ast.ImportFrom) and not node.level and node.module:
imported.add(node.module.split(".", 1)[0])
return imported

30
scripts/check_silent_failures.py
View File

@ -24,8 +24,13 @@ from __future__ import annotations
import ast
import logging
from pathlib import Path
import sys
from typing import TYPE_CHECKING
from scripts._silent_cli import run_hook
if TYPE_CHECKING:
from pathlib import Path
_logger = logging.getLogger(__name__)
@ -61,7 +66,9 @@ def _is_silent_exit(node: ast.AST) -> bool:
if not is_exit or len(node.args) != 1:
return False
arg = node.args[0]
return isinstance(arg, ast.Constant) and arg.value == 0
return isinstance(arg, ast.Constant) and (
isinstance(arg.value, int) and not arg.value
)
def _find_silent_exits(tree: ast.AST) -> list[ast.Call]:
@ -177,24 +184,7 @@ def _check_source(path: Path, source: str) -> list[str]:
def main() -> int:
"""Return 1 if any file swallows an exception silently, else 0."""
problems: list[str] = []
for filename in sys.argv[1:]:
path = Path(filename)
try:
source = path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
_logger.warning(
"Could not read %s (%s) — reporting it as a problem rather than "
"skipping it, since its handlers cannot be checked",
path,
exc,
)
problems.append(f"{path}: could not read ({exc})")
continue
problems.extend(_check_source(path, source))
for problem in problems:
sys.stdout.write(f"{problem}\n")
return 1 if problems else 0
return run_hook(_check_source)
if __name__ == "__main__":

28
scripts/check_silent_failures_dart.py
View File

@ -30,9 +30,14 @@ not a skip, exactly like the Python hook's SyntaxError branch.
from __future__ import annotations
import logging
from pathlib import Path
import re
import sys
from typing import TYPE_CHECKING
from scripts._silent_cli import run_hook
if TYPE_CHECKING:
from pathlib import Path
_logger = logging.getLogger(__name__)
@ -113,7 +118,7 @@ def _block_after(source: str, start: int) -> tuple[str, int] | None:
depth += 1
elif source[idx] == "}":
depth -= 1
if depth == 0:
if not depth:
return source[brace : idx + 1], idx
return None
@ -187,24 +192,7 @@ def _check_source(path: Path, source: str) -> list[str]:
def main() -> int:
"""Return 1 if any Dart file swallows an error silently, else 0."""
problems: list[str] = []
for filename in sys.argv[1:]:
path = Path(filename)
try:
source = path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
_logger.warning(
"Could not read %s (%s) — reporting it as a problem rather "
"than skipping it, since its handlers cannot be checked",
path,
exc,
)
problems.append(f"{path}: could not read ({exc})")
continue
problems.extend(_check_source(path, source))
for problem in problems:
sys.stdout.write(f"{problem}\n")
return 1 if problems else 0
return run_hook(_check_source)
if __name__ == "__main__":

18
scripts/demo_workout_handoff.py
View File

@ -29,20 +29,16 @@ also still works -- but the hatch above is the one this demo is testing.
One thing a green run does NOT prove: under the demo's local grab gatelock's
watchdog early-returns and never re-takes the grab, so the 1000ms steal-back
that makes ``_recovery.stop()`` load-bearing in production is not exercised.
that makes ``recovery.stop()`` load-bearing in production is not exercised.
python3 scripts/demo_workout_handoff.py
python3 -m scripts.demo_workout_handoff # from the repo root
"""
from __future__ import annotations
import logging
from pathlib import Path
import sys
import tkinter as tk
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from gatelock import GrabPolicy, LockConfig, LockWindow
from screen_locker._workout_app import ProcessHooks, workout_app_binary
@ -59,6 +55,7 @@ class _DemoHooks:
self.session: WorkoutSession | None = None
def build_surface(self, parent: tk.Misc, _surface: object) -> None:
"""Draw the status line and the release/reacquire buttons."""
tk.Label(
parent,
text="DEMO LOCK — local grab, VT switching still enabled",
@ -115,11 +112,14 @@ class _DemoHooks:
if self.lock is not None: # pragma: no cover
self.lock.close()
def teardown_surface(self, _surface: object) -> None: ...
def teardown_surface(self, _surface: object) -> None:
"""Nothing to tear down in the demo."""
def on_focus_ready(self, _surface: object) -> None: ...
def on_focus_ready(self, _surface: object) -> None:
"""The demo needs no focus handling."""
def on_close(self) -> None: ...
def on_close(self) -> None:
"""The demo needs no close handling."""
def main() -> int:

14
scripts/preview_shutdown.py
View File

@ -98,7 +98,6 @@ def _simulate(args: argparse.Namespace, tmp: Path) -> None:
if args.rest_day:
eve = datetime.combine(day - timedelta(days=1), time(12)).astimezone()
_say(f" declare: {declare(day, now=eve, path=rest_file).reason}")
rest = is_rest_day(day, rest_file)
done = {n for n in (args.done or "").split(",") if n}
flat = {e.name: e.name in done for e in _earned.flat_earners()}
first = stamp(day, args.first_done) if args.first_done else None
@ -109,7 +108,7 @@ def _simulate(args: argparse.Namespace, tmp: Path) -> None:
day,
flat,
day_credit_count(log, day),
rest_day=rest,
rest_day=is_rest_day(day, rest_file),
first_done=min(times, default=None),
sick_day=sick_pool(day, sick_day=args.sick_day),
)
@ -130,9 +129,8 @@ def _sequence(args: argparse.Namespace, tmp: Path) -> None:
vars(module)["GRACE_STATE_FILE"] = tmp / "grace_floor.json"
use_temp_pool(tmp)
done = {e.name: False for e in _earned.flat_earners()}
start = derive(DayInputs(day, done))
sim = SimShutdown(start.minutes)
_say(f"[sequence] {day} start {hhmm(start.minutes)} (nothing done, not sick)")
sim = SimShutdown(derive(DayInputs(day, done)).minutes)
_say(f"[sequence] {day} start {hhmm(sim.config[0])} (nothing done, not sick)")
workouts, first, went_late = 0, None, False
# The real target reads today's real ledgers; the sim's are ``done``.
vars(_sick_floor)["sick_target"] = lambda d, _log: (
@ -177,12 +175,10 @@ def main() -> int:
parser.add_argument("--sequence", help="name@HH:MM,... live passes in order")
parser.add_argument("--log", type=Path, default=_LOG, help="log.json (read-only)")
args = parser.parse_args()
ladder = getattr(earned_time, "on_ladder", None)
_say(f"earned_time from {Path(earned_time.__file__).parent}")
if args.day is None:
today = date.fromisoformat(today_str())
on = ladder(today) if ladder else False
_say(f" ladder in force today: {on}")
_say(f" ladder in force today: {earned_time.on_ladder(today)}")
live = read_shutdown_config(SHUTDOWN_CONFIG_FILE)
_say(
f" live config: {tuple(hhmm(m) for m in live) if live else 'unreadable'}"
@ -195,7 +191,7 @@ def main() -> int:
_report(f"today {label}", derive(replace(inputs, sick_day=sick)))
return 0
day = date.fromisoformat(args.day)
_say(f" ladder in force on {day}: {ladder(day) if ladder else False}")
_say(f" ladder in force on {day}: {earned_time.on_ladder(day)}")
with tempfile.TemporaryDirectory(prefix="preview_shutdown_") as raw:
if args.sequence:
_sequence(args, Path(raw))

2
scripts/refresh_refactor_todo.py
View File

@ -110,7 +110,7 @@ def commits_last_year(path: Path) -> int:
text=True,
check=False,
)
if result.returncode != 0:
if result.returncode:
_logger.warning(
"git log failed for %s (%s) — counting 0 commits, so its ROI "
"rank will read lower than it should",

4
scripts/rekey_log_entry.py
View File

@ -71,7 +71,7 @@ def main() -> int:
log_path = Path(args.log_file)
try:
log = json.loads(log_path.read_text())
log = json.loads(log_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
_logger.exception("Cannot read %s", log_path)
_report(f"ERROR: cannot read {log_path}: {exc}")
@ -83,7 +83,7 @@ def main() -> int:
_report(f"ERROR: {exc} — nothing changed")
return 1
try:
log_path.write_text(json.dumps(log, indent=2))
log_path.write_text(json.dumps(log, indent=2), encoding="utf-8")
except OSError as exc:
_logger.exception("Could not write %s", log_path)
_report(f"ERROR: could not write {log_path}: {exc}")

17
scripts/split_dart_test.py
View File

@ -33,7 +33,7 @@ def top_level_groups(lines: list[str]) -> list[int]:
def split(path: Path, specs: list[tuple[Path, int]]) -> None:
"""Write each spec's group range to its own file, trimming the source."""
lines = path.read_text().splitlines(keepends=True)
lines = path.read_text(encoding="utf-8").splitlines(keepends=True)
groups = top_level_groups(lines)
main_idx = next(i for i, ln in enumerate(lines) if ln.startswith("void main("))
header = "".join(lines[:main_idx])
@ -47,16 +47,17 @@ def split(path: Path, specs: list[tuple[Path, int]]) -> None:
)
trailer = "".join(lines[main_end + 1 :])
bounds = [i for _, i in specs] + [len(groups)]
for (out, start), end in zip(specs, bounds[1:], strict=False):
first = groups[start]
ends = [i for _, i in specs[1:]] + [len(groups)]
for (out, start), end in zip(specs, ends, strict=True):
last = groups[end] if end < len(groups) else main_end
body = "".join(lines[first:last]).rstrip("\n")
out.write_text(f"{header}void main() {{\n{preamble}{body}\n}}\n{trailer}")
body = "".join(lines[groups[start] : last]).rstrip("\n")
out.write_text(
f"{header}void main() {{\n{preamble}{body}\n}}\n{trailer}", encoding="utf-8"
)
keep = groups[specs[0][1]]
path.write_text(
"".join(lines[:keep]).rstrip("\n") + "\n}\n" + trailer,
"".join(lines[: groups[specs[0][1]]]).rstrip("\n") + "\n}\n" + trailer,
encoding="utf-8",
)

2
scripts/verify_workout_credited.py
View File

@ -63,7 +63,7 @@ def main() -> int:
return 1
try:
log = json.loads(log_path.read_text())
log = json.loads(log_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
_logger.exception(
"Could not read %s — cannot tell whether the workout was "