"""RunnerUp run auto-verification via ADB (file-based path + shared validation). File-based (no root, works over WiFi): reads per-activity TCX files that RunnerUp's File Synchronizer writes to ``/sdcard/Documents/RunnerUp/``. Root DB fallback lives in ``_runnerup_db.py``. """ from __future__ import annotations import logging 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 _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", } # TCX uses sport name strings; map back to integer codes for unified validation. class RunnerUpVerificationMixin( RunnerUpDbMixin, RunnerUpTcxMixin, RunnerUpBackfillMixin, ): """Mixin providing RunnerUp-based workout verification via ADB.""" # ------------------------------------------------------------------ # File-based path (no root required) # ------------------------------------------------------------------ def _has_runnerup_source(self) -> bool: """Whether any RunnerUp export source is reachable right now. The WebDAV drop directory on this disk counts as much as an adb-visible phone: both hold the same TCX files. """ return self._has_local_runnerup_dir() or self._has_adb_device() @staticmethod def _has_local_runnerup_dir() -> bool: return any(d.is_dir() for d in RUNNERUP_WEBDAV_DIRS) @staticmethod def _find_local_runnerup_exports_for_date(date_str: str) -> list[str]: """TCX files for ``date_str`` in the WebDAV drop directories, sorted.""" found: list[str] = [] for local_dir in RUNNERUP_WEBDAV_DIRS: if not local_dir.is_dir(): continue for path in sorted(local_dir.glob("*.tcx")): if date_str in path.name and str(path) not in found: found.append(str(path)) return found def _find_runnerup_exports_for_date(self, date_str: str) -> list[str]: """Return paths of RunnerUp TCX exports for the given date, or empty list. Local WebDAV copies come first (no phone needed), then the phone's export directories over adb when a device is attached. Args: date_str: ISO date string in ``YYYY-MM-DD`` format matched against TCX filenames (``RunnerUp_YYYY-MM-DD-HH-MM-SS_xxx.tcx``). """ found = self._find_local_runnerup_exports_for_date(date_str) if not self._has_adb_device(): return found for dirpath in RUNNERUP_EXPORT_DIRS: ok, out = self._run_adb(["shell", "ls", dirpath]) if not ok or not out.strip(): continue for raw in out.strip().splitlines(): name = raw.strip() if date_str in name and name.endswith(".tcx"): remote = f"{dirpath}/{name}" if remote not in found: found.append(remote) return found def _verify_runnerup_via_files(self) -> tuple[str, str] | None: """Try to verify today's run via TCX export files. Returns ``(status, message)`` if a today's file was found (even if it fails validation), or ``None`` if no today's file exists at all (caller should try the root DB path instead). """ today = today_str() exports = self._find_runnerup_exports_for_date(today) if not exports: _logger.warning( "No RunnerUp TCX export found for %s in %s — a run today will " "not be verified unless RunnerUp's File Synchronizer " "auto-export is enabled and the app stays enabled long enough " "to sync (check it isn't disabled by focus mode).", today, RUNNERUP_EXPORT_DIRS, ) 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", ) 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``. """ 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: _logger.info( "RunnerUp distance %.2f km is under the %.1f km minimum but " "within %.0f%% GPS tolerance — accepted.", 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", ) # ------------------------------------------------------------------ # Main entry point # ------------------------------------------------------------------ def _verify_runnerup_workout(self) -> tuple[str, str]: """Verify today's run via RunnerUp. Tries TCX file exports first (no root, works over WiFi); falls back to root DB pull if no today's files are found. Status values: ``verified | not_verified | no_phone | too_short | clock_tampered``. """ skew_ok, skew_msg = check_clock_skew() if not skew_ok: return "clock_tampered", skew_msg if not self._has_runnerup_source(): return ( "no_phone", "Phone not connected — plug in via USB or enable wireless ADB", ) # Path 1: file-based (no root needed; WebDAV copies first, then adb). file_result = self._verify_runnerup_via_files() if file_result is not None: _logger.info("RunnerUp file-based result: %s", file_result[0]) return file_result # Path 2: root DB pull (fallback when no export files found yet). # On a non-rooted device this path cannot succeed; _verify_runnerup_via_db # reports that plainly rather than as a generic failure. Needs adb. if not self._has_adb_device(): return ( "not_verified", ( f"No RunnerUp export for today in {RUNNERUP_WEBDAV_DIRS} " "and the phone is not adb-reachable" ), ) return self._verify_runnerup_via_db()