"""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 from datetime import datetime, timedelta, timezone import json import logging from pathlib import Path import shutil import tempfile from typing import Any import xml.etree.ElementTree as ET from gatelock.log_integrity import compute_entry_hmac from screen_locker._constants import ( MIN_RUN_DISTANCE_KM, MIN_RUN_DURATION_MINUTES, RUNNERUP_ACCEPTED_SPORTS, RUNNERUP_EXPORT_DIRS, ) from screen_locker._runnerup_db import RunnerUpDbMixin from screen_locker._time_check import check_clock_skew from screen_locker._weekly_check import COUNTED_WORKOUT_TYPES _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. _TCX_SPORT_TO_INT: dict[str, int] = {v: k for k, v in _SPORT_NAMES.items()} # TCX XML namespace used by Garmin/RunnerUp. _TCX_NS = "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2" class RunnerUpVerificationMixin(RunnerUpDbMixin): """Mixin providing RunnerUp-based workout verification via ADB.""" # ------------------------------------------------------------------ # File-based path (no root required) # ------------------------------------------------------------------ def _find_runnerup_exports_for_date(self, date_str: str) -> list[str]: """Return adb paths of RunnerUp TCX exports for the given date, or empty list. 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: list[str] = [] 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 _pull_and_parse_tcx(self, remote_path: str) -> dict[str, Any] | None: """Pull a remote TCX file and parse it. Returns activity dict or None.""" tmp_dir = tempfile.mkdtemp(prefix="runnerup_tcx_") local_path = str(Path(tmp_dir) / "activity.tcx") try: ok, _ = self._run_adb(["pull", remote_path, local_path]) if not ok or not Path(local_path).exists(): _logger.info("Failed to pull TCX file: %s", remote_path) return None return self._parse_tcx(local_path) finally: shutil.rmtree(tmp_dir, ignore_errors=True) def _parse_tcx(self, tcx_path: str) -> dict[str, Any] | None: """Parse a local TCX file and return activity summary dict. Sums ``TotalTimeSeconds`` and ``DistanceMeters`` across all Laps so multi-segment runs (pause/resume) are counted in full. """ try: tree = ET.parse(tcx_path) except ET.ParseError as exc: _logger.info("TCX parse error in %s: %s", tcx_path, exc) return None root = tree.getroot() activity = root.find(f".//{{{_TCX_NS}}}Activity") if activity is None: _logger.info("No Activity element in TCX file") return None sport_str = activity.get("Sport", "") sport_int = _TCX_SPORT_TO_INT.get(sport_str, -1) total_seconds = 0.0 total_distance = 0.0 for lap in activity.findall(f"{{{_TCX_NS}}}Lap"): t_elem = lap.find(f"{{{_TCX_NS}}}TotalTimeSeconds") d_elem = lap.find(f"{{{_TCX_NS}}}DistanceMeters") if t_elem is not None and t_elem.text: total_seconds += float(t_elem.text) if d_elem is not None and d_elem.text: total_distance += float(d_elem.text) return { "sport": sport_int, "duration_seconds": int(total_seconds), "distance_m": total_distance, } 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 = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d") exports = self._find_runnerup_exports_for_date(today) if not exports: return None # Try each file; return the best result (verified > validation error). best: tuple[str, str] | None = None for remote in exports: data = self._pull_and_parse_tcx(remote) if data is None: continue status, msg = self._validate_runnerup_data(data) if status == "verified": return status, msg if best is None: best = (status, msg) # All files found but none passed validation. return best or ( "not_verified", "RunnerUp TCX export found but could not be read", ) def _try_fill_runnerup_for_date(self, date_str: str, logs: dict[str, Any]) -> bool: """Try to fill one date gap from RunnerUp TCX exports, mutating logs in-place. Returns True if a verified entry was written for ``date_str``. """ existing = logs.get(date_str, {}) if isinstance(existing, dict): wtype = existing.get("workout_data", {}).get("type", "") if wtype in COUNTED_WORKOUT_TYPES: return False for remote in self._find_runnerup_exports_for_date(date_str): data = self._pull_and_parse_tcx(remote) if data is None: continue status, msg = self._validate_runnerup_data(data) if status != "verified": continue entry: dict[str, Any] = { "timestamp": datetime.now(tz=timezone.utc).isoformat(), "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), }, } signature = compute_entry_hmac(entry) if signature is not None: entry["hmac"] = signature logs[date_str] = entry _logger.info("Auto-filled RunnerUp entry for %s: %s", date_str, msg) return True return False def _scan_and_fill_week_runnerup(self, log_file: Path) -> int: """Scan the current ISO week for RunnerUp TCX gaps and fill them. Returns the count of newly filled entries (0 if phone not connected). """ if not self._has_adb_device(): _logger.info( "Phone not connected; skipping auto-scan for past RunnerUp exports." ) return 0 now = datetime.now(tz=timezone.utc).astimezone() today = now.date() week_start = today - timedelta(days=today.weekday()) try: with log_file.open() as f: logs: dict[str, Any] = json.load(f) except (OSError, json.JSONDecodeError): logs = {} filled = 0 current = week_start while current <= today: if self._try_fill_runnerup_for_date(current.strftime("%Y-%m-%d"), logs): filled += 1 current += timedelta(days=1) if filled > 0: try: with log_file.open("w") as f: json.dump(logs, f, indent=2) except OSError as exc: _logger.warning("Failed to write workout log after scan: %s", exc) return 0 return filled # ------------------------------------------------------------------ # Shared validation # ------------------------------------------------------------------ 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"] if sport not in RUNNERUP_ACCEPTED_SPORTS: sport_name = _SPORT_NAMES.get(sport, f"unknown({sport})") return ( "wrong_sport", f"Activity type '{sport_name}' doesn't count as a qualifying run", ) duration_min = data["duration_seconds"] / 60 if duration_min < MIN_RUN_DURATION_MINUTES: msg = ( f"Run was {duration_min:.0f} min — need {MIN_RUN_DURATION_MINUTES}+ min" ) return "too_short", msg distance_km = data["distance_m"] / 1000 if distance_km < MIN_RUN_DISTANCE_KM: msg = f"Run was {distance_km:.1f} km — need {MIN_RUN_DISTANCE_KM:.0f}+ km" return "too_short", msg 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 | wrong_sport | clock_tampered``. """ skew_ok, skew_msg = check_clock_skew() if not skew_ok: return "clock_tampered", skew_msg if not self._has_adb_device(): return ( "no_phone", "Phone not connected — plug in via USB or enable wireless ADB", ) # Path 1: file-based (no root needed). 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). _logger.info("No TCX exports found today; trying root DB pull...") return self._verify_runnerup_via_db()