diff --git a/README.md b/README.md index 6a2b0df..9bd1bd3 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,18 @@ so a run finished after 20:00 pushes the hour later without the phone being plugged in; the per-minute shutdown check then lifts the night lockdown on its own once the hour is past "now" (digital-wellbeing's check script). +Runs and walks are credited by ONE rule, whatever the TCX sport or the +`_Walking` file-name suffix (RunnerUp's files and the workout app's alike): a +day's activities are summed, and the day counts once their distance reaches +3.5 km (5% GPS tolerance) OR their *moving* time reaches the shared duration +bar (30 min minus the hidden leeway). Both are summed over trackpoint pairs +seen within 30 s of each other and travelled at no more than 25 km/h (faster +is a vehicle or a GPS jump: no time, no distance); moving time also needs +0.5 m/s or more. Pauses, standing still, dropouts and files without +trackpoints earn nothing (the lap totals are never used). The reward is the usual one +workout credit per day (`runnerup_verified:`); the log's `activity` +field is display only. Code: `screen_locker/_walk_tcx.py`. + **Retiring RunnerUp.** A week after the PC first credits a run recorded by the workout app (TCX author `kuhy workout_app`), `runnerup-retire.timer` (daily, retries until the phone is attached) backs `/sdcard/Documents/RunnerUp` up to diff --git a/screen_locker/_constants.py b/screen_locker/_constants.py index 8143f3c..7eaacb4 100644 --- a/screen_locker/_constants.py +++ b/screen_locker/_constants.py @@ -73,11 +73,11 @@ WORKOUT_DURATION_ACCEPT_MINUTES = ( ) RUNNERUP_PACKAGES = ("org.runnerup", "org.runnerup.free") RUNNERUP_DB_SDCARD_TMP = "/sdcard/.runnerup_tmp_verification.db" -# A run qualifies on EITHER criterion — they are an OR, not an AND. A short -# but long-lasting run counts, and so does a fast 5 km that takes under the -# duration bar. Do not reinstate an AND here: it would reject exactly the -# slow-run case this pair exists to allow. The duration arm is the shared -# MIN_WORKOUT_DURATION_MINUTES above; only the distance arm is run-specific. +# A day's activity (run or walk alike) qualifies on EITHER criterion — they are +# an OR, not an AND. A short but long-lasting session counts, and so does a +# fast 5 km that takes under the duration bar. Do not reinstate an AND here: +# it would reject exactly the slow case this pair exists to allow. The +# duration arm is the shared MIN_WORKOUT_DURATION_MINUTES above (moving time). MIN_RUN_DISTANCE_KM = 3.5 # inclusive: "as low as 3.5 km" # GPS distance tracking undershoots on tight turns/tree cover/urban canyons; # a run within 5% of the minimum is accepted rather than rejected on @@ -105,10 +105,16 @@ SHUTDOWN_BASE_FILE = Path(__file__).resolve().parent / "shutdown_base.json" REST_DAY_FILE = Path(__file__).resolve().parent / "rest_days.json" REST_DAYS_PER_ISO_WEEK = 2 GRACE_STATE_FILE = Path(__file__).resolve().parent / "grace_floor.json" -# A day's RunnerUp walks (commute legs) count as the workout once their -# MOVING time reaches WORKOUT_DURATION_ACCEPT_MINUTES (the shared bar above); -# moving = trackpoint pairs at walking speed. -WALK_SPEED_MPS = (0.5, 3.0) +# Every GPS activity (run or walk, whatever its TCX sport) is judged the same +# way: a day's activities count as the workout once their summed distance +# reaches MIN_RUN_DISTANCE_KM OR their summed MOVING time reaches +# WORKOUT_DURATION_ACCEPT_MINUTES. Moving = trackpoint pairs seen within +# WALK_MAX_GAP_SECONDS and travelled at least this fast (slower is standing +# still). A pair faster than MAX_MOVING_SPEED_MPS (25 km/h, the app's running +# jump filter) is a vehicle or a GPS jump and counts neither moving time NOR +# distance, for every activity. The Dart app mirrors the bounds. +MIN_MOVING_SPEED_MPS = 0.5 +MAX_MOVING_SPEED_MPS = 25 / 3.6 WALK_MAX_GAP_SECONDS = 30 # Workout records this PC must keep deleted in the sync store, pushed as # tombstones on every tick (see _sync_tombstones for why once is not enough). diff --git a/screen_locker/_runnerup_backfill.py b/screen_locker/_runnerup_backfill.py index 2b546cf..6bc2ae3 100644 --- a/screen_locker/_runnerup_backfill.py +++ b/screen_locker/_runnerup_backfill.py @@ -15,7 +15,7 @@ import logging from typing import TYPE_CHECKING, Any from screen_locker._log_mixin import write_signed_entry -from screen_locker._walk_tcx import walk_day, walk_shortfall +from screen_locker._walk_tcx import activity_day, activity_shortfall if TYPE_CHECKING: from collections.abc import Callable @@ -72,66 +72,47 @@ class RunnerUpBackfillMixin: Returns True if a new verified entry was appended for ``date_str``. """ - walks: list[dict[str, Any]] = [] - for remote in self._find_runnerup_exports_for_date(date_str): - data = self._pull_and_parse_tcx(remote) - if data is None: - continue - if data.get("walking"): - # Never one walk at a time: a day's walks are summed below. - walks.append(data) - continue - status, msg = self._validate_runnerup_data(data) - if status != "verified": - _logger.warning( - "RunnerUp export %s for %s did not qualify (%s): %s", - remote, - date_str, - status, - msg, - ) - continue - workout_data = { - "type": "runnerup_verified", - "source": f"Auto-scanned: {msg}", - "distance_km": round(data["distance_m"] / 1000, 2), - "duration_minutes": round(data["duration_seconds"] / 60, 1), - "completed_at": data.get("ended_at"), - } - return _record(log_file, date_str, workout_data, on_ingested) - return self._try_fill_walks(date_str, walks, log_file, on_ingested) + activities = [ + data + for remote in self._find_runnerup_exports_for_date(date_str) + if (data := self._pull_and_parse_tcx(remote)) is not None + ] + return self._try_fill_day(date_str, activities, log_file, on_ingested) @staticmethod - def _try_fill_walks( + def _try_fill_day( date_str: str, - walks: list[dict[str, Any]], + activities: list[dict[str, Any]], log_file: Path, on_ingested: OnIngestedCallback | None, ) -> bool: - """Append the day's walking as its RunnerUp workout, once it adds up. + """Append the day's activity as its RunnerUp workout, once it adds up. - Same type and so the same ``runnerup_verified:{date}`` slot as a run: - a walk and a run on one day are one RunnerUp credit, exactly as two - runs already were. + One rule for runs and walks alike (see :mod:`screen_locker._walk_tcx`): + the day's activities are summed, never judged one file at a time. They + share the ``runnerup_verified:{date}`` slot, so a walk and a run on + one day are one RunnerUp credit. ``activity: walking`` is kept for the + phone's history display only; it never decides credit. """ - if not walks: + if not activities: return False - day = walk_day(walks) + day = activity_day(activities) if not day.qualifies: _logger.warning( - "RunnerUp walks for %s do not count yet: %s", + "RunnerUp activity for %s does not count yet: %s", date_str, - walk_shortfall(day), + activity_shortfall(day), ) return False - workout_data = { + workout_data: dict[str, Any] = { "type": "runnerup_verified", - "activity": "walking", "source": f"Auto-scanned: {day.message()}", "distance_km": round(day.distance_km, 2), "duration_minutes": round(day.moving_minutes, 1), "completed_at": day.completed_at, } + if day.all_walks: + workout_data["activity"] = "walking" return _record(log_file, date_str, workout_data, on_ingested) def _scan_and_fill_week_runnerup( diff --git a/screen_locker/_runnerup_verification.py b/screen_locker/_runnerup_verification.py index c7a40f2..9b65e04 100644 --- a/screen_locker/_runnerup_verification.py +++ b/screen_locker/_runnerup_verification.py @@ -12,31 +12,26 @@ from typing import Any from screen_locker._constants import ( MIN_RUN_DISTANCE_KM, - MIN_WORKOUT_DURATION_MINUTES, RUNNERUP_DISTANCE_TOLERANCE, RUNNERUP_EXPORT_DIRS, RUNNERUP_WEBDAV_DIRS, - WORKOUT_DURATION_ACCEPT_MINUTES, ) from screen_locker._day import today_str from screen_locker._runnerup_backfill import RunnerUpBackfillMixin from screen_locker._runnerup_db import RunnerUpDbMixin from screen_locker._runnerup_tcx import RunnerUpTcxMixin from screen_locker._time_check import check_clock_skew -from screen_locker._walk_tcx import walk_day, walk_shortfall +from screen_locker._walk_tcx import ( + SPORT_NAMES, + activity_day, + activity_shortfall, + distance_qualifies, +) _logger = logging.getLogger(__name__) -_SPORT_NAMES: dict[int, str] = { - 0: "Running", - 1: "Biking", - 2: "Other", - 3: "Orienteering", - 4: "Walking", - 5: "Treadmill", - 6: "Gym", - 7: "Stationary Bike", -} +# Kept for importers; the labels live next to the one crediting rule. +_SPORT_NAMES = SPORT_NAMES # TCX uses sport name strings; map back to integer codes for unified validation. @@ -121,73 +116,46 @@ class RunnerUpVerificationMixin( ) return None - # Try each file; return the best result (verified > validation error). - # Walks never qualify one file at a time: they are summed below. - best: tuple[str, str] | None = None - walks: list[dict[str, Any]] = [] - for remote in exports: - data = self._pull_and_parse_tcx(remote) - if data is None: - continue - if data.get("walking"): - walks.append(data) - continue - status, msg = self._validate_runnerup_data(data) - if status == "verified": - return status, msg - if best is None: - best = (status, msg) - if walks: - day = walk_day(walks) - if day.qualifies: - return "verified", day.message() - best = best or ("too_short", walk_shortfall(day)) - - # All files found but none passed validation. - return best or ( - "not_verified", - "RunnerUp TCX export found but could not be read", - ) + # One rule for the whole day, whatever the sport or file name: the + # day's activities are summed (see _walk_tcx), never judged one file + # at a time. + activities = [ + data + for remote in exports + if (data := self._pull_and_parse_tcx(remote)) is not None + ] + if not activities: + return ( + "not_verified", + "RunnerUp TCX export found but could not be read", + ) + day = activity_day(activities) + if day.qualifies: + return "verified", day.message() + return "too_short", activity_shortfall(day) def _validate_runnerup_data(self, data: dict[str, Any]) -> tuple[str, str]: """Validate a RunnerUp activity against configured thresholds. Returns ``(status, message)`` following the same contract as - ``PhoneVerificationMixin._verify_phone_workout``. + ``PhoneVerificationMixin._verify_phone_workout``. The one rule is + :func:`screen_locker._walk_tcx.activity_day`; a dict without + ``moving_seconds`` (the root-DB path) is judged by its duration. """ - sport = data["sport"] - - duration_min = data["duration_seconds"] / 60 - distance_km = data["distance_m"] / 1000 - - # Either criterion qualifies the run on its own (see _constants.py). - # Distance gets the GPS tolerance (a measurement correction); duration - # gets the shared hidden leeway (a deliberate concession). Different - # reasons, so they stay separate constants. - distance_ok = ( - distance_km * (1 + RUNNERUP_DISTANCE_TOLERANCE) >= MIN_RUN_DISTANCE_KM - ) - duration_ok = duration_min >= WORKOUT_DURATION_ACCEPT_MINUTES - if not distance_ok and not duration_ok: - msg = ( - f"Run was {distance_km:.1f} km / {duration_min:.0f} min — need " - f"{MIN_RUN_DISTANCE_KM:g}+ km or {MIN_WORKOUT_DURATION_MINUTES}+ min" - ) - return "too_short", msg - if distance_ok and distance_km < MIN_RUN_DISTANCE_KM: + day = activity_day([data]) + if not day.qualifies: + return "too_short", activity_shortfall(day) + if day.distance_km < MIN_RUN_DISTANCE_KM and distance_qualifies( + day.distance_km + ): _logger.info( "RunnerUp distance %.2f km is under the %.1f km minimum but " "within %.0f%% GPS tolerance — accepted.", - distance_km, + day.distance_km, MIN_RUN_DISTANCE_KM, RUNNERUP_DISTANCE_TOLERANCE * 100, ) - - sport_name = _SPORT_NAMES.get(sport, str(sport)) - return ( - "verified", - f"{sport_name}: {distance_km:.1f} km in {duration_min:.0f} min", - ) + return "verified", day.message() # ------------------------------------------------------------------ # Main entry point diff --git a/screen_locker/_walk_tcx.py b/screen_locker/_walk_tcx.py index 13c924e..fb6fe22 100644 --- a/screen_locker/_walk_tcx.py +++ b/screen_locker/_walk_tcx.py @@ -1,23 +1,35 @@ -"""Walking in RunnerUp TCX exports: which files are walks, and their moving time. +"""One crediting rule for every GPS activity, from its RunnerUp-style TCX export. -A day's walking counts as the workout once its walks -- usually separate -commute legs, one file each -- add up to the shared workout duration bar -(``WORKOUT_DURATION_ACCEPT_MINUTES``) of *moving* time. +A day's activities -- runs and walks, RunnerUp's files and the workout app's +alike -- are summed, and the day counts as the workout once EITHER -**Which files are walks.** RunnerUp's WebDAV and File synchronisers both -export with ``ExportOptions.getDefault()``, whose TCX writer knows only -Running/Biking and writes ``Sport="Other"`` for a walk. The sport survives -only in the file name (``Sport.TapiriikType()``): ``..._Walking.tcx``. The -Strava export variant writes ``Sport="Walking"``, so that counts too. -``Other`` alone is not a walk: it is every sport RunnerUp cannot name. +* their distance reaches ``MIN_RUN_DISTANCE_KM`` (within the GPS tolerance), OR +* their *moving* time reaches the shared duration bar + (``WORKOUT_DURATION_ACCEPT_MINUTES``). + +The TCX sport and the ``_Walking`` file-name suffix never change the outcome; +they only decide the label a message and the log's ``activity`` field carry +(the phone's history shows it). + +**Which files are walks (label only).** RunnerUp's WebDAV and File +synchronisers both export with ``ExportOptions.getDefault()``, whose TCX writer +knows only Running/Biking and writes ``Sport="Other"`` for a walk. The sport +survives only in the file name (``Sport.TapiriikType()``): ``..._Walking.tcx``. +The Strava export variant writes ``Sport="Walking"``, so that counts too. **Moving time, not elapsed.** A lap's ``TotalTimeSeconds`` is wall-clock: one real lap reads 101 min for 1 km of standing about. Moving time is summed over -consecutive trackpoint pairs whose speed is walking speed -(``WALK_SPEED_MPS``): standing still is too slow, a tram or a bus too fast. -A pair further apart than ``WALK_MAX_GAP_SECONDS`` is a GPS dropout whose -motion is unknown and earns nothing -- the walk is never credited for time -nobody saw. +consecutive trackpoint pairs travelled at least at ``MIN_MOVING_SPEED_MPS``: +standing still is too slow. A pair further apart than ``WALK_MAX_GAP_SECONDS`` +is a GPS dropout whose motion is unknown and earns nothing -- an activity is +never credited for time nobody saw, so a pause cannot buy a run its duration. +A pair faster than ``MAX_MOVING_SPEED_MPS`` (25 km/h) is a vehicle or a GPS +jump: it counts neither moving time nor distance, so a bus ride earns nothing. +The day's distance is the sum of the surviving pairs (``track_distance_m``), +never the lap's ``DistanceMeters`` total, and an export with no trackpoints +cannot be speed-checked, so it credits nothing. Only the root-DB fallback, +which has no trackpoints at all, is judged by its recorded distance and +duration. """ from __future__ import annotations @@ -29,9 +41,12 @@ import logging from typing import TYPE_CHECKING, Any from screen_locker._constants import ( + MAX_MOVING_SPEED_MPS, + MIN_MOVING_SPEED_MPS, + MIN_RUN_DISTANCE_KM, MIN_WORKOUT_DURATION_MINUTES, + RUNNERUP_DISTANCE_TOLERANCE, WALK_MAX_GAP_SECONDS, - WALK_SPEED_MPS, WORKOUT_DURATION_ACCEPT_MINUTES, ) @@ -44,6 +59,18 @@ _logger = logging.getLogger(__name__) _TCX_NS = "{http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2}" _WALKING = "Walking" +# RunnerUp's sport codes, for labels only (the label never decides credit). +SPORT_NAMES: dict[int, str] = { + 0: "Running", + 1: "Biking", + 2: "Other", + 3: "Orienteering", + 4: "Walking", + 5: "Treadmill", + 6: "Gym", + 7: "Stationary Bike", +} + def is_walking_export(name: str, sport: str) -> bool: """Whether a TCX export is a walk, from its file name or its Sport tag.""" @@ -74,72 +101,135 @@ def _trackpoints(activity: ET.Element) -> list[tuple[float, float]]: def track_summary(activity: ET.Element) -> dict[str, Any]: - """Moving seconds and end time of one activity, from its trackpoints.""" + """Moving seconds, credited distance and end time, from the trackpoints. + + A pair counts when its points are within ``WALK_MAX_GAP_SECONDS`` and it + was not travelled faster than ``MAX_MOVING_SPEED_MPS``. It then adds its + metres to ``track_distance_m``, and its seconds to ``moving_seconds`` too + when it was travelled at least at ``MIN_MOVING_SPEED_MPS``. + """ points = _trackpoints(activity) - slow, fast = WALK_SPEED_MPS moving = 0.0 + metres = 0.0 for (t0, d0), (t1, d1) in itertools.pairwise(points): gap = t1 - t0 - if 0 < gap <= WALK_MAX_GAP_SECONDS and slow <= (d1 - d0) / gap <= fast: + if not 0 < gap <= WALK_MAX_GAP_SECONDS: + continue + speed = (d1 - d0) / gap + if speed > MAX_MOVING_SPEED_MPS: + continue + metres += max(d1 - d0, 0.0) + if speed >= MIN_MOVING_SPEED_MPS: moving += gap return { "activity_id": activity.findtext(f"{_TCX_NS}Id") or "", "moving_seconds": moving, + "track_distance_m": metres, "ended_at": points[-1][0] if points else None, } +def distance_qualifies(distance_km: float) -> bool: + """Whether a distance meets the minimum, with the GPS tolerance.""" + return distance_km * (1 + RUNNERUP_DISTANCE_TOLERANCE) >= MIN_RUN_DISTANCE_KM + + +def moving_qualifies(moving_minutes: float) -> bool: + """Whether moving time meets the shared duration bar (leeway included).""" + return moving_minutes >= WORKOUT_DURATION_ACCEPT_MINUTES + + +def _distance_metres(activity: dict[str, Any]) -> float: + """Credited metres; the root-DB path has no trackpoints, so its total stands in.""" + return float(activity.get("track_distance_m", activity.get("distance_m", 0.0))) + + +def _moving_seconds(activity: dict[str, Any]) -> float: + """Moving seconds; the root-DB path has no trackpoints, so wall-clock stands in.""" + return float(activity.get("moving_seconds", activity.get("duration_seconds", 0.0))) + + @dataclass(frozen=True) -class WalkDay: - """A day's walks, summed. +class ActivityDay: + """A day's activities, summed. Attributes: - legs: Distinct walks (deduplicated by the activity's ``Id``: the - WebDAV copy and the phone's copy are the same walk). + legs: Distinct activities (deduplicated by the activity's ``Id``: the + WebDAV copy and the phone's copy are the same activity). moving_minutes: Their summed moving time. distance_km: Their summed distance. - completed_at: Unix time the walk that crossed the bar ended -- when - the workout was actually done -- or ``None`` if it never did. + completed_at: Unix time the activity that took the day over the bar + ended -- when the workout was actually done -- or ``None`` if the + day never got there (or the activity recorded no end). + all_walks: Whether every activity was a walk (label only). + sport: The sport code when the day is a single non-walk activity. """ legs: int moving_minutes: float distance_km: float completed_at: float | None + all_walks: bool = False + sport: int = -1 @property def qualifies(self) -> bool: - """Whether the day's walking counts as its workout.""" - return self.moving_minutes >= WORKOUT_DURATION_ACCEPT_MINUTES + """Whether the day counts as its workout: distance OR moving time.""" + return distance_qualifies(self.distance_km) or moving_qualifies( + self.moving_minutes + ) def message(self) -> str: """The verification line, in the RunnerUp verifier's own style.""" + if self.all_walks: + return ( + f"Walking: {self.legs} walk(s), {self.distance_km:.1f} km, " + f"{self.moving_minutes:.0f} min moving" + ) + if self.legs == 1: + name = SPORT_NAMES.get(self.sport, str(self.sport)) + return f"{name}: {self.distance_km:.1f} km in {self.moving_minutes:.0f} min" return ( - f"Walking: {self.legs} walk(s), {self.distance_km:.1f} km, " + f"Activity: {self.legs} activities, {self.distance_km:.1f} km, " f"{self.moving_minutes:.0f} min moving" ) -def walk_shortfall(day: WalkDay) -> str: - """Why a day's walking did not count yet.""" - return f"{day.message()} -- need {MIN_WORKOUT_DURATION_MINUTES}+ min moving" +def activity_shortfall(day: ActivityDay) -> str: + """Why a day's activity did not count yet.""" + return ( + f"{day.message()} -- need {MIN_RUN_DISTANCE_KM:g}+ km or " + f"{MIN_WORKOUT_DURATION_MINUTES}+ min moving" + ) -def walk_day(walks: Iterable[dict[str, Any]]) -> WalkDay: - """Sum parsed walk exports (``_parse_tcx`` dicts) into one :class:`WalkDay`.""" +def activity_day(activities: Iterable[dict[str, Any]]) -> ActivityDay: + """Sum parsed exports (``parse_tcx_file`` dicts) into one :class:`ActivityDay`.""" unique: dict[str, dict[str, Any]] = {} - for walk in walks: - unique.setdefault(str(walk.get("activity_id") or id(walk)), walk) - ordered = sorted(unique.values(), key=lambda w: w.get("ended_at") or 0.0) + for activity in activities: + unique.setdefault(str(activity.get("activity_id") or id(activity)), activity) + ordered = sorted(unique.values(), key=lambda a: a.get("ended_at") or 0.0) moving = 0.0 + metres = 0.0 completed_at: float | None = None - for walk in ordered: - moving += float(walk.get("moving_seconds", 0.0)) - if completed_at is None and moving / 60 >= WORKOUT_DURATION_ACCEPT_MINUTES: - completed_at = walk.get("ended_at") - return WalkDay( + for activity in ordered: + moving += _moving_seconds(activity) + metres += _distance_metres(activity) + if completed_at is None and ( + moving_qualifies(moving / 60) or distance_qualifies(metres / 1000) + ): + completed_at = activity.get("ended_at") + return ActivityDay( legs=len(ordered), moving_minutes=moving / 60, - distance_km=sum(float(w.get("distance_m", 0.0)) for w in ordered) / 1000, + distance_km=metres / 1000, completed_at=completed_at, + all_walks=bool(ordered) and all(a.get("walking") for a in ordered), + sport=int(ordered[0].get("sport", -1)) if ordered else -1, ) + + +# Back-compat names: the walk-only rule these replaced. +WalkDay = ActivityDay +walk_day = activity_day +walk_shortfall = activity_shortfall diff --git a/screen_locker/_workout_ledger.py b/screen_locker/_workout_ledger.py index 411f875..1df2546 100644 --- a/screen_locker/_workout_ledger.py +++ b/screen_locker/_workout_ledger.py @@ -4,8 +4,8 @@ Every other earner publishes its credits as HMAC-signed rows in a ledger that ``earned_time`` reads (``first_credit_at``, ``credit_units``). This is the workout's: ``~/.local/share/workout_locker/ledger.json``, one row per -* workout credit slot in ``log.json`` -- RunnerUp (a run, or the day's walks - summed), StrongLifts (phone or PC session), manual, any lock-screen or +* workout credit slot in ``log.json`` -- RunnerUp (the day's runs and walks, + summed by one rule), StrongLifts (phone or PC session), manual, any lock-screen or verify path -- appended by the one log write chokepoint (``_log_mixin.write_signed_entry``), so no path can credit without it; and * declared rest day -- appended when it is declared. diff --git a/screen_locker/tests/_activity_helpers.py b/screen_locker/tests/_activity_helpers.py new file mode 100644 index 0000000..fbf1a4c --- /dev/null +++ b/screen_locker/tests/_activity_helpers.py @@ -0,0 +1,96 @@ +"""Helpers for the unified-activity tests: TCX builder and a phone-less verifier.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import TYPE_CHECKING +from unittest.mock import patch + +from screen_locker._runnerup_verification import RunnerUpVerificationMixin + +if TYPE_CHECKING: + from pathlib import Path + +_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" +DAY = "2026-10-10" + + +def tcx( + sport: str, + start: str, + *, + moving_min: float, + metres: float, + pause_min: float = 0.0, +) -> str: + """A RunnerUp-shaped export: ``moving_min`` of steady motion covering ``metres``. + + A ``pause_min`` pause sits in the middle of the track as ONE gap between + two trackpoints (the clock runs, the distance does not), so the lap's + wall-clock time is ``moving_min + pause_min``. + """ + begin = datetime.fromisoformat(f"{DAY}T{start}:00+00:00") + speed = metres / (moving_min * 60) + total = int(moving_min * 60) + half = total // 2 // 5 * 5 + pause = int(pause_min * 60) + points = "" + for s in range(0, total + 1, 5): + offset = s + (pause if s > half else 0) + points += ( + f"{s * speed:.1f}" + ) + return ( + f'' + f'{DAY}T{start}' + f"{total + pause}" + f"{metres}" + f"{points}" + "" + ) + + +class Verifier(RunnerUpVerificationMixin): + """The real mixin reading one temporary drop directory and no phone.""" + + def __init__(self, folder: Path) -> None: + self.folder = folder + + def _has_adb_device(self) -> bool: + return False + + def _run_adb(self, args: list[str]) -> tuple[bool, str]: + return False, "" + + def _find_runnerup_exports_for_date(self, date_str: str) -> list[str]: + return sorted(str(p) for p in self.folder.glob(f"*{date_str}*.tcx")) + + +def drop(folder: Path, stamp: str, suffix: str, body: str) -> None: + """Write a RunnerUp-named export for the test day.""" + (folder / f"RunnerUp_Pixel_6a_{DAY}-{stamp}{suffix}.tcx").write_text(body) + + +def verify(verifier: Verifier) -> tuple[str, str] | None: + """The file-based verdict for the test day.""" + with patch("screen_locker._runnerup_verification.today_str", return_value=DAY): + return verifier._verify_runnerup_via_files() + + +def tcx_points(sport: str, start: str, points: list[tuple[int, float]]) -> str: + """An export whose track is exactly ``(seconds after start, cumulative metres)``.""" + begin = datetime.fromisoformat(f"{DAY}T{start}:00+00:00") + body = "".join( + f"" + f"{m:.1f}" + for s, m in points + ) + return ( + f'' + f'{DAY}T{start}' + f"{points[-1][0]}" + f"{points[-1][1]}" + f"{body}" + "" + ) diff --git a/screen_locker/tests/test_runnerup_walks.py b/screen_locker/tests/test_runnerup_walks.py index 5c768b2..6c9d494 100644 --- a/screen_locker/tests/test_runnerup_walks.py +++ b/screen_locker/tests/test_runnerup_walks.py @@ -116,26 +116,31 @@ class TestVerifyToday: drop(verifier.folder, "08-00-00_Walking", tcx("Other", 60, "08:00", 0.1)) assert self._verify(verifier) == ( "too_short", - "Walking: 1 walk(s), 0.4 km, 0 min moving -- need 30+ min moving", + "Walking: 1 walk(s), 0.4 km, 0 min moving -- need 3.5+ km or 30+ min moving", ) - def test_a_run_still_verifies_beside_short_walks(self, verifier: Verifier) -> None: + def test_a_run_beside_short_walks_is_one_summed_day( + self, verifier: Verifier + ) -> None: """A long enough run verifies even when the day also has short walks.""" drop(verifier.folder, "07-00-00_Walking", tcx("Other", 10, "07:00")) drop(verifier.folder, "18-00-00_Running", tcx("Running", 45, "18:00", 2.5)) result = self._verify(verifier) assert result is not None assert result[0] == "verified" - assert result[1].startswith("Running") + assert result[1].startswith("Activity: 2 activities") - def test_a_short_run_outranks_the_walk_shortfall(self, verifier: Verifier) -> None: - """A too-short run is the reported shortfall rather than the walks'.""" + def test_a_short_run_and_short_walks_share_one_shortfall( + self, verifier: Verifier + ) -> None: + """A too-short run and short walks are summed into one shortfall.""" drop(verifier.folder, "07-00-00_Walking", tcx("Other", 10, "07:00")) drop(verifier.folder, "18-00-00_Running", tcx("Running", 5, "18:00", 2.5)) result = self._verify(verifier) assert result is not None assert result[0] == "too_short" - assert result[1].startswith("Run was") + assert result[1].startswith("Activity: 2 activities") + assert "need 3.5+ km or 30+ min moving" in result[1] class TestBackfill: @@ -174,7 +179,7 @@ class TestBackfill: log = tmp_path / "log.json" with caplog.at_level(logging.WARNING): assert verifier._try_fill_runnerup_for_date(_DAY, log) is False - assert "do not count yet" in caplog.text + assert "does not count yet" in caplog.text assert not log.exists() def test_a_run_records_its_end_time( diff --git a/screen_locker/tests/test_runnerup_webdav_source.py b/screen_locker/tests/test_runnerup_webdav_source.py index 7360c1a..9757c30 100644 --- a/screen_locker/tests/test_runnerup_webdav_source.py +++ b/screen_locker/tests/test_runnerup_webdav_source.py @@ -12,25 +12,14 @@ from __future__ import annotations from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch +from screen_locker.tests._activity_helpers import tcx from screen_locker.tests.conftest import create_locker if TYPE_CHECKING: from pathlib import Path -_TCX = """\ - - - - - - 4200.0 - 5800.0 - - - - -""" +# A 5.8 km / 70 min run WITH its trackpoints: lap totals alone credit nothing. +_TCX = tcx("Running", "18:58", moving_min=70, metres=5800) def _webdav_tree(tmp_path: Path) -> tuple[Path, Path]: diff --git a/screen_locker/tests/test_tcx_unified_day.py b/screen_locker/tests/test_tcx_unified_day.py new file mode 100644 index 0000000..8c2048c --- /dev/null +++ b/screen_locker/tests/test_tcx_unified_day.py @@ -0,0 +1,236 @@ +"""The unified rule sums a day's activities; the root DB and the pure aggregate.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import json +import logging +from typing import TYPE_CHECKING + +import pytest + +from screen_locker._walk_tcx import ( + ActivityDay, + activity_day, + activity_shortfall, + walk_day, +) +from screen_locker.tests._activity_helpers import ( + DAY, + Verifier, + drop, + tcx, + tcx_points, + verify, +) + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture(name="verifier") +def verifier_fixture(tmp_path: Path) -> Verifier: + """A Verifier reading a temporary drop directory.""" + return Verifier(tmp_path) + + +class TestDaySums: + """Time AND distance add up across a day's files, whatever their sport.""" + + def test_two_15_min_sessions_make_the_bar(self, verifier: Verifier) -> None: + """A run and a walk of 15 moving minutes each credit together.""" + drop( + verifier.folder, + "07-00-00", + "_Running", + tcx("Running", "07:00", moving_min=15, metres=900), + ) + drop( + verifier.folder, + "18-00-00", + "_Walking", + tcx("Other", "18:00", moving_min=15, metres=900), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "verified" + + def test_two_short_distances_make_the_distance_bar( + self, verifier: Verifier + ) -> None: + """Distance sums per day too: 2 km + 1.6 km is 3.6 km.""" + drop( + verifier.folder, + "07-00-00", + "_Running", + tcx("Running", "07:00", moving_min=10, metres=2000), + ) + drop( + verifier.folder, + "18-00-00", + "_Walking", + tcx("Other", "18:00", moving_min=10, metres=1600), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "verified" + + def test_the_same_activity_from_two_sources_counts_once( + self, verifier: Verifier + ) -> None: + """The WebDAV and phone copies of one export are one activity.""" + body = tcx("Running", "07:00", moving_min=15, metres=900) + drop(verifier.folder, "07-00-00", "_Running", body) + drop(verifier.folder, "07-00-01", "_Running", body) + result = verify(verifier) + assert result is not None + assert result[0] == "too_short" + + def test_a_short_day_reports_what_is_missing( + self, verifier: Verifier, caplog: pytest.LogCaptureFixture + ) -> None: + """The shortfall names both bars and what the day had.""" + drop( + verifier.folder, + "07-00-00", + "_Running", + tcx("Running", "07:00", moving_min=10, metres=800), + ) + with caplog.at_level(logging.WARNING): + result = verify(verifier) + assert result is not None + assert result[0] == "too_short" + assert "need 3.5+ km or 30+ min moving" in result[1] + + def test_the_completion_time_is_the_activity_that_crossed_the_bar( + self, verifier: Verifier, tmp_path: Path + ) -> None: + """completed_at is when the day's total crossed the bar, not its last file.""" + drop( + verifier.folder, + "07-00-00", + "_Running", + tcx("Running", "07:00", moving_min=20, metres=1200), + ) + drop( + verifier.folder, + "12-00-00", + "_Walking", + tcx("Other", "12:00", moving_min=10, metres=600), + ) + drop( + verifier.folder, + "18-00-00", + "_Walking", + tcx("Other", "18:00", moving_min=10, metres=600), + ) + log = tmp_path / "log.json" + assert verifier._try_fill_runnerup_for_date(DAY, log) is True + (entry,) = json.loads(log.read_text())[DAY] + crossed = datetime(2026, 10, 10, 12, 10, tzinfo=UTC).timestamp() + assert entry["workout_data"]["completed_at"] == crossed + + +class TestRootDbFallback: + """The root DB path has no trackpoints: wall-clock is all it knows.""" + + def test_a_db_row_is_judged_by_its_duration(self, verifier: Verifier) -> None: + """No moving_seconds in the dict: the recorded duration stands in.""" + row = {"sport": 0, "duration_seconds": 31 * 60, "distance_m": 2000.0} + assert verifier._validate_runnerup_data(row)[0] == "verified" + row["duration_seconds"] = 10 * 60 + assert verifier._validate_runnerup_data(row)[0] == "too_short" + + def test_sport_code_does_not_change_the_verdict(self, verifier: Verifier) -> None: + """Every RunnerUp sport code gets the same answer for the same numbers.""" + base = {"duration_seconds": 31 * 60, "distance_m": 2000.0} + verdicts = { + verifier._validate_runnerup_data({**base, "sport": code})[0] + for code in (-1, 0, 1, 2, 3, 4, 5, 6, 7) + } + assert verdicts == {"verified"} + + +class TestActivityDay: + """The pure day aggregate.""" + + def test_distance_gets_the_gps_tolerance(self) -> None: + """3.4 km is within 5% of the 3.5 km minimum; 3.3 km is not.""" + near = activity_day([{"distance_m": 3400.0, "moving_seconds": 0.0}]) + assert near.qualifies + far = activity_day([{"distance_m": 3300.0, "moving_seconds": 0.0}]) + assert not far.qualifies + + def test_the_alias_still_imports(self) -> None: + """walk_day is the old name of activity_day.""" + assert walk_day is activity_day + + def test_message_styles(self) -> None: + """A lone run reads like a run, walks like walks, a mix like activities.""" + run = {"sport": 0, "distance_m": 4200.0, "moving_seconds": 1560.0} + walk = {"walking": True, "distance_m": 2300.0, "moving_seconds": 1680.0} + assert activity_day([run]).message() == "Running: 4.2 km in 26 min" + assert activity_day([walk]).message() == ( + "Walking: 1 walk(s), 2.3 km, 28 min moving" + ) + assert activity_day([run, walk]).message() == ( + "Activity: 2 activities, 6.5 km, 54 min moving" + ) + assert activity_day([{"sport": 99}]).message().startswith("99:") + + def test_shortfall_text(self) -> None: + """The shortfall names both ways to qualify.""" + day: ActivityDay = activity_day([{"distance_m": 100.0, "moving_seconds": 60.0}]) + assert activity_shortfall(day).endswith("need 3.5+ km or 30+ min moving") + + +class TestLapOnlyExport: + """An export with laps but no trackpoints has no moving time to credit.""" + + def test_wall_clock_alone_credits_nothing(self, verifier: Verifier) -> None: + """40 min on the lap clock, 1 km, no trackpoints: too short.""" + body = ( + '' + "2026-10-10T08:002400" + "1000" + "" + ) + drop(verifier.folder, "08-00-00", "_Running", body) + result = verify(verifier) + assert result is not None + assert result[0] == "too_short" + + +class TestSpeedCap: + """Pairs faster than 25 km/h count neither moving time nor distance.""" + + def test_a_car_trip_credits_nothing(self, verifier: Verifier) -> None: + """10 km at 40 km/h (15 min) is a vehicle, not a workout.""" + drop( + verifier.folder, + "08-00-00", + "_Walking", + tcx("Walking", "08:00", moving_min=15, metres=10000), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "too_short" + + def test_one_gps_jump_drops_only_its_pair(self, verifier: Verifier) -> None: + """A 60 km/h jump inside a 3 m/s run loses that pair and nothing else.""" + points = [(s, s * 3.0 + (83.0 if s > 600 else 0.0)) for s in range(0, 1201, 5)] + drop( + verifier.folder, + "08-00-00", + "_Running", + tcx_points("Running", "08:00", points), + ) + (path,) = verifier._find_runnerup_exports_for_date(DAY) + data = verifier._pull_and_parse_tcx(path) + assert data is not None + assert data["moving_seconds"] == pytest.approx(1195) + assert data["track_distance_m"] == pytest.approx(3585) + result = verify(verifier) + assert result is not None + assert result[0] == "verified" diff --git a/screen_locker/tests/test_tcx_unified_rule.py b/screen_locker/tests/test_tcx_unified_rule.py new file mode 100644 index 0000000..11d4835 --- /dev/null +++ b/screen_locker/tests/test_tcx_unified_rule.py @@ -0,0 +1,158 @@ +"""One crediting rule for every GPS activity: sport and file name never matter. + +A day's activities (runs, walks, RunnerUp's files and the workout app's alike) +are summed and credit once their distance reaches ``MIN_RUN_DISTANCE_KM`` (with +the GPS tolerance) OR their *moving* time reaches the shared duration bar. +Wall-clock time never counts: a pause or a standing-still stretch earns +nothing. Everything runs on temporary directories. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from screen_locker.tests._activity_helpers import DAY, Verifier, drop, tcx, verify + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture(name="verifier") +def verifier_fixture(tmp_path: Path) -> Verifier: + """A Verifier reading a temporary drop directory.""" + return Verifier(tmp_path) + + +class TestRequiredCases: + """The four cases kuhy named when he unified runs and walks.""" + + def test_a_walking_file_of_3_6_km_in_40_min_is_credited( + self, verifier: Verifier + ) -> None: + """(a) A Walking-sport export at 3.6 km / 40 min credits.""" + drop( + verifier.folder, + "08-00-00", + "_Walking", + tcx("Walking", "08:00", moving_min=40, metres=3600), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "verified" + + def test_a_walking_file_of_3_6_km_in_18_min_is_credited_by_distance( + self, verifier: Verifier + ) -> None: + """(a) Distance alone credits a walk, as it always did a run.""" + drop( + verifier.folder, + "08-00-00", + "_Walking", + tcx("Walking", "08:00", moving_min=18, metres=3600), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "verified" + + def test_a_running_file_with_31_min_moving_and_2_km_is_credited( + self, verifier: Verifier + ) -> None: + """(b) Moving time alone credits a run.""" + drop( + verifier.folder, + "18-00-00", + "_Running", + tcx("Running", "18:00", moving_min=31, metres=2000), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "verified" + + def test_a_pause_does_not_buy_a_run_its_duration(self, verifier: Verifier) -> None: + """(c) 24 min moving + 10 min pause = 34 min wall-clock: not credited. + + (24 not 25: the duration bar is 30 minus the 5 minute leeway, so a + full 25 minutes of motion DOES credit.) + """ + drop( + verifier.folder, + "18-00-00", + "_Running", + tcx("Running", "18:00", moving_min=24, metres=1600, pause_min=10), + ) + result = verify(verifier) + assert result is not None + assert result[0] == "too_short" + + def test_a_pause_does_not_buy_a_run_its_log_entry( + self, verifier: Verifier, tmp_path: Path + ) -> None: + """(c) The backfill path agrees: nothing is logged.""" + drop( + verifier.folder, + "18-00-00", + "_Running", + tcx("Running", "18:00", moving_min=24, metres=1600, pause_min=10), + ) + log = tmp_path / "log.json" + assert verifier._try_fill_runnerup_for_date(DAY, log) is False + assert not log.exists() + + +_TRACKS = { + "qualifying_by_time": {"moving_min": 31, "metres": 2000}, + "qualifying_by_distance": {"moving_min": 18, "metres": 3600}, + "paused_short": {"moving_min": 24, "metres": 1600, "pause_min": 10}, + "short": {"moving_min": 10, "metres": 800}, +} +_DRESSINGS = [ + ("Running", "_Running"), + ("Walking", "_Walking"), + ("Other", "_Walking"), + ("Other", ""), + ("Biking", ""), +] + + +class TestSportAndNameNeverMatter: + """(d) The same track gives the same outcome under any sport or file name.""" + + @pytest.mark.parametrize("track", sorted(_TRACKS)) + def test_status_and_log_are_identical(self, track: str, tmp_path: Path) -> None: + """Every dressing of one track verifies, logs and credits alike.""" + outcomes = set() + for index, (sport, suffix) in enumerate(_DRESSINGS): + folder = tmp_path / f"d{index}" + folder.mkdir() + verifier = Verifier(folder) + drop(folder, "08-00-00", suffix, tcx(sport, "08:00", **_TRACKS[track])) + status = verify(verifier) + assert status is not None + log = folder / "log.json" + filled = verifier._try_fill_runnerup_for_date(DAY, log) + credited = ( + json.loads(log.read_text())[DAY][0]["workout_data"]["completed_at"] + if filled + else None + ) + outcomes.add((status[0], filled, credited)) + assert len(outcomes) == 1 + + def test_the_activity_label_is_kept_but_never_decides( + self, verifier: Verifier, tmp_path: Path + ) -> None: + """A walk keeps ``activity: walking`` in the log for the phone's display.""" + drop( + verifier.folder, + "08-00-00", + "_Walking", + tcx("Other", "08:00", moving_min=31, metres=2000), + ) + log = tmp_path / "log.json" + assert verifier._try_fill_runnerup_for_date(DAY, log) is True + (entry,) = json.loads(log.read_text())[DAY] + assert entry["workout_data"]["activity"] == "walking" + assert entry["workout_data"]["type"] == "runnerup_verified" diff --git a/screen_locker/tests/test_walk_tcx.py b/screen_locker/tests/test_walk_tcx.py index af8f7c2..b0fcb87 100644 --- a/screen_locker/tests/test_walk_tcx.py +++ b/screen_locker/tests/test_walk_tcx.py @@ -72,10 +72,12 @@ class TestTrackSummary: assert summary["activity_id"] == "A1" assert summary["ended_at"] == (_START + timedelta(seconds=600)).timestamp() - def test_standing_and_vehicles_are_not_moving(self) -> None: - """Standing-still and vehicle speeds add no moving time.""" + def test_standing_is_not_moving_but_a_run_is(self) -> None: + """Standing still adds no moving time; any speed above it does (runs).""" assert not track_summary(walk(600, speed=0.1))["moving_seconds"] - assert not track_summary(walk(600, speed=8.0))["moving_seconds"] + assert track_summary(walk(600, speed=4.5))["moving_seconds"] == ( + pytest.approx(600) + ) def test_a_gps_dropout_earns_nothing(self) -> None: """Two points 100 s apart at walking pace: motion unseen, not counted.""" @@ -101,7 +103,12 @@ class TestTrackSummary: def test_no_points_has_no_end(self) -> None: """A track with no points has zero moving time and no end.""" summary = track_summary(activity([])) - assert summary == {"activity_id": "A1", "moving_seconds": 0.0, "ended_at": None} + assert summary == { + "activity_id": "A1", + "moving_seconds": 0.0, + "track_distance_m": 0.0, + "ended_at": None, + } def leg( @@ -110,6 +117,7 @@ def leg( """Build a per-walk summary dict like track_summary produces.""" return { "activity_id": ident, + "walking": True, "moving_seconds": minutes * 60, "ended_at": ended, "distance_m": metres, @@ -148,7 +156,7 @@ class TestWalkDay: def test_shortfall_says_what_is_missing(self) -> None: """The shortfall text states how much moving time is still needed.""" text = walk_shortfall(walk_day([leg("a", 20, 1.0)])) - assert text.endswith("need 30+ min moving") + assert text.endswith("need 3.5+ km or 30+ min moving") def test_walks_share_the_one_workout_accept_bar(self) -> None: # Same advertised-30 / accept-25 bar as every other workout type. diff --git a/stronglift_replacement/workout_app/DOCS-workout-app-context.md b/stronglift_replacement/workout_app/DOCS-workout-app-context.md index 598bd30..fe2e5f8 100644 --- a/stronglift_replacement/workout_app/DOCS-workout-app-context.md +++ b/stronglift_replacement/workout_app/DOCS-workout-app-context.md @@ -25,8 +25,11 @@ Implementing design_v2.md improvements. LOCAL time; ``/`