diff --git a/screen_locker/_constants.py b/screen_locker/_constants.py index 58bd9af..1478eaf 100644 --- a/screen_locker/_constants.py +++ b/screen_locker/_constants.py @@ -43,6 +43,12 @@ WORKOUT_APP_JSON_REMOTES = ( # Port the workout app's HTTP server listens on (no ADB/developer-options needed). WORKOUT_HTTP_PORT = 8765 MIN_WORKOUT_DURATION_MINUTES = 60 +RUNNERUP_PACKAGES = ("org.runnerup", "org.runnerup.free") +RUNNERUP_DB_SDCARD_TMP = "/sdcard/.runnerup_tmp_verification.db" +MIN_RUN_DURATION_MINUTES = 30 +MIN_RUN_DISTANCE_KM = 5.0 +RUNNERUP_ACCEPTED_SPORTS: frozenset[int] = frozenset({0, 3, 5}) +# 0=RUNNING, 3=ORIENTEERING, 5=TREADMILL MAX_CLOCK_SKEW_SECONDS = 300 # 5 minutes max time skew from NTP EARLY_BIRD_START_HOUR = 5 EARLY_BIRD_END_HOUR = 8 diff --git a/screen_locker/_runnerup_verification.py b/screen_locker/_runnerup_verification.py new file mode 100644 index 0000000..9e2cbc2 --- /dev/null +++ b/screen_locker/_runnerup_verification.py @@ -0,0 +1,225 @@ +"""RunnerUp run auto-verification via ADB SQLite DB pull. + +Pulls RunnerUp's private database directly from a rooted device, then +queries it locally. No user interaction is required — the whole pipeline +runs in the background just like ``_phone_verification.py``. +""" + +from __future__ import annotations + +import logging +import os +import shutil +import sqlite3 +import tempfile +import time +from typing import Any + +from screen_locker._constants import ( + MIN_RUN_DISTANCE_KM, + MIN_RUN_DURATION_MINUTES, + RUNNERUP_ACCEPTED_SPORTS, + RUNNERUP_DB_SDCARD_TMP, + RUNNERUP_PACKAGES, +) +from screen_locker._time_check import check_clock_skew + +_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", +} + + +class RunnerUpVerificationMixin: + """Mixin providing RunnerUp-based workout verification via ADB DB pull.""" + + def _find_runnerup_package(self) -> str | None: + """Return the first installed RunnerUp package name, or None.""" + for pkg in RUNNERUP_PACKAGES: + ok, out = self._adb_shell(f"pm list packages {pkg}") + if ok and pkg in out: + return pkg + return None + + def _pull_runnerup_db(self) -> str | None: + """Pull RunnerUp's SQLite DB from the device to a local temp file. + + Copies the DB and any WAL/SHM sidecar files via root shell to + ``/sdcard`` (accessible without root by adb pull), then pulls them + locally. WAL files must travel with the main DB so that + ``PRAGMA wal_checkpoint`` can merge in-flight writes. + + Returns the local DB path on success, or ``None`` on any failure. + """ + pkg = self._find_runnerup_package() + if pkg is None: + _logger.info("RunnerUp not installed (tried %s)", RUNNERUP_PACKAGES) + return None + + db_device = f"/data/data/{pkg}/databases/runnerup.db" + tmp_dir = tempfile.mkdtemp(prefix="runnerup_verify_") + local_db = os.path.join(tmp_dir, "runnerup.db") + + # Copy the main DB to sdcard where adb pull can reach it (root needed). + ok, err = self._adb_shell( + f"cp {db_device} {RUNNERUP_DB_SDCARD_TMP}", + root=True, + ) + if not ok: + _logger.info("Failed to copy RunnerUp DB to sdcard: %s", err) + shutil.rmtree(tmp_dir, ignore_errors=True) + return None + + # Copy WAL and SHM sidecars if they exist; ignore failure (they may not). + for suffix in ("-wal", "-shm"): + self._adb_shell( + f"test -f {db_device}{suffix} " + f"&& cp {db_device}{suffix} {RUNNERUP_DB_SDCARD_TMP}{suffix} " + f"|| true", + root=True, + ) + + # Pull main DB. + ok, _ = self._run_adb(["pull", RUNNERUP_DB_SDCARD_TMP, local_db]) + if not ok: + _logger.info("adb pull of RunnerUp DB failed") + self._cleanup_runnerup_sdcard() + shutil.rmtree(tmp_dir, ignore_errors=True) + return None + + # Pull sidecars (best-effort; sqlite3 tolerates missing ones). + for suffix in ("-wal", "-shm"): + self._run_adb( + ["pull", f"{RUNNERUP_DB_SDCARD_TMP}{suffix}", f"{local_db}{suffix}"] + ) + + self._cleanup_runnerup_sdcard() + return local_db + + def _cleanup_runnerup_sdcard(self) -> None: + """Remove temporary RunnerUp DB files from the sdcard.""" + for suffix in ("", "-wal", "-shm"): + self._adb_shell( + f"test -f {RUNNERUP_DB_SDCARD_TMP}{suffix} " + f"&& rm {RUNNERUP_DB_SDCARD_TMP}{suffix} || true", + root=True, + ) + + def _query_todays_run(self, db_path: str) -> dict[str, Any] | None: + """Query the pulled RunnerUp DB for today's most recent activity. + + Runs ``PRAGMA wal_checkpoint`` first so that any uncommitted WAL + entries are visible to the query (important for runs just finished). + + Returns a dict with ``distance_m``, ``duration_seconds``, and + ``sport`` on success; ``None`` if no matching row exists. + """ + # Build today's epoch window in local time (RunnerUp stores seconds). + local_midnight = time.mktime(time.localtime()[:3] + (0, 0, 0, 0, 0, -1)) + local_end = local_midnight + 86400 + + try: + with sqlite3.connect(db_path, timeout=5) as conn: + conn.execute("PRAGMA wal_checkpoint(PASSIVE)") + cursor = conn.execute( + """ + SELECT start_time, distance, time, type + FROM activity + WHERE deleted = 0 + AND start_time >= ? + AND start_time < ? + ORDER BY start_time DESC + LIMIT 1 + """, + (local_midnight, local_end), + ) + row = cursor.fetchone() + except sqlite3.Error as exc: + _logger.info("RunnerUp DB query failed: %s", exc) + return None + + if row is None: + return None + + start_time, distance_m, duration_seconds, sport = row + return { + "start_time": int(start_time or 0), + "distance_m": float(distance_m or 0), + "duration_seconds": int(duration_seconds or 0), + "sport": int(sport or 0), + } + + 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: + return ( + "too_short", + f"Run was {duration_min:.0f} min — need at least {MIN_RUN_DURATION_MINUTES} min", + ) + + distance_km = data["distance_m"] / 1000 + if distance_km < MIN_RUN_DISTANCE_KM: + return ( + "too_short", + f"Run was {distance_km:.1f} km — need at least {MIN_RUN_DISTANCE_KM:.0f} km", + ) + + sport_name = _SPORT_NAMES.get(sport, str(sport)) + return ( + "verified", + f"{sport_name}: {distance_km:.1f} km in {duration_min:.0f} min", + ) + + def _verify_runnerup_workout(self) -> tuple[str, str]: + """Verify today's run via RunnerUp DB pull. + + Entry point mirroring ``PhoneVerificationMixin._verify_phone_workout``. + 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 ADB to verify RunnerUp run", + ) + + db_path = self._pull_runnerup_db() + if db_path is None: + return "not_verified", "Could not retrieve RunnerUp database from phone" + + try: + run_data = self._query_todays_run(db_path) + finally: + shutil.rmtree(os.path.dirname(db_path), ignore_errors=True) + + if run_data is None: + return "not_verified", "No RunnerUp activity found for today" + + return self._validate_runnerup_data(run_data) diff --git a/screen_locker/_ui_flows.py b/screen_locker/_ui_flows.py index 8533111..6a511cc 100644 --- a/screen_locker/_ui_flows.py +++ b/screen_locker/_ui_flows.py @@ -84,25 +84,59 @@ class UIFlowsMixin: "Actually do the full workout, don't just\n" "spam through the exercises.", ) - elif status in ("stale", "no_exercises"): - self._show_retry_and_sick( - f"❌ {message}\n\nReason: {status}", - ) elif status == "clock_tampered": self._show_retry_and_sick( f"❌ {message}\n\n" "System clock appears to be manipulated.\n" "Fix your system time and try again.", ) - elif status == "not_verified": - self._show_retry_and_sick( - f"❌ {message}\n\n" - "StrongLifts shows no workout today.\n" - "Go do your workout first!", + elif status in ("stale", "no_exercises", "not_verified"): + # Try RunnerUp before showing failure — user may have run instead of lifted. + self._start_runnerup_fallback( + lambda: self._show_retry_and_sick( + f"❌ {message}\n\n" + "Neither StrongLifts nor RunnerUp found a workout today.\n" + "Go do your workout first!", + ) ) else: - # no_phone or error — penalty timer, then retry+sick screen - self._show_phone_penalty(message) + # no_phone or error — try RunnerUp first, then penalty timer. + self._start_runnerup_fallback(lambda: self._show_phone_penalty(message)) + + def _start_runnerup_fallback( + self, on_failure: "Callable[[], None]" + ) -> None: + """Check RunnerUp as fallback after phone check fails. + + Shows a waiting screen, runs the check in a background thread, then + either unlocks (run verified) or calls ``on_failure``. + """ + self.clear_container() + self._label("Checking RunnerUp...", font_size=36, color="#ffaa00", pady=30) + self._text("Looking for today's run in RunnerUp...", font_size=18) + executor = ThreadPoolExecutor(max_workers=1) + self._runnerup_future = executor.submit(self._verify_runnerup_workout) + executor.shutdown(wait=False) + self._runnerup_on_failure = on_failure + self._poll_runnerup_fallback() + + def _poll_runnerup_fallback(self) -> None: + """Poll the RunnerUp background check and route to result handler.""" + if self._runnerup_future is not None and self._runnerup_future.done(): + status, message = self._runnerup_future.result() + if status == "verified": + self.workout_data["type"] = "runnerup_verified" + self.workout_data["source"] = message + self.clear_container() + self._label("✓ Run Verified!", font_size=42, color="#00cc44", pady=30) + self._text(message, font_size=20, color="#aaffaa") + self._text("Unlocking...", font_size=18, color="#888888") + unlock_delay = 1500 if self.demo_mode else 2000 + self.root.after(unlock_delay, self.unlock_screen) + else: + self._runnerup_on_failure() + else: + self.root.after(500, self._poll_runnerup_fallback) def ask_if_sick(self) -> None: """Display the structured sick-day justification dialog.""" diff --git a/screen_locker/_weekly_check.py b/screen_locker/_weekly_check.py index 0c99a37..303d885 100644 --- a/screen_locker/_weekly_check.py +++ b/screen_locker/_weekly_check.py @@ -26,8 +26,13 @@ WEEKLY_WORKOUT_MINIMUM: int = 4 # Python weekday(): Mon=0, Tue=1, Wed=2, Thu=3, Fri=4, Sat=5, Sun=6 _RELAXED_WEEKDAYS: frozenset[int] = frozenset({1, 2, 3}) # Tue, Wed, Thu -# Only phone-verified workouts count toward the weekly minimum. -_COUNTED_WORKOUT_TYPES: frozenset[str] = frozenset({"phone_verified"}) +# Workout types that count toward the weekly minimum *and* earn the +# shutdown-time bonus (see screen_lock._try_adjust_shutdown_for_workout). +# Exported (no leading underscore) so screen_lock.py can share this single +# source of truth instead of duplicating the type check. +COUNTED_WORKOUT_TYPES: frozenset[str] = frozenset( + {"phone_verified", "runnerup_manual", "runnerup_verified"}, +) def is_relaxed_day(*, today: datetime | None = None) -> bool: @@ -86,7 +91,7 @@ def count_weekly_workouts( if not isinstance(entry, dict): continue wtype = entry.get("workout_data", {}).get("type", "") - if wtype in _COUNTED_WORKOUT_TYPES: + if wtype in COUNTED_WORKOUT_TYPES: count += 1 return count diff --git a/screen_locker/screen_lock.py b/screen_locker/screen_lock.py index 05e42c0..c86126f 100755 --- a/screen_locker/screen_lock.py +++ b/screen_locker/screen_lock.py @@ -32,6 +32,7 @@ from screen_locker._constants import ( ) from screen_locker._early_bird import EarlyBirdMixin from screen_locker._phone_verification import PhoneVerificationMixin +from screen_locker._runnerup_verification import RunnerUpVerificationMixin from screen_locker._shutdown import ShutdownMixin from screen_locker._sick_dialog import SickDialogMixin from screen_locker._ui_flows import UIFlowsMixin @@ -39,6 +40,7 @@ from screen_locker._ui_flows_relaxed import UIFlowsRelaxedMixin from screen_locker._ui_widgets import UIWidgetsMixin from screen_locker._wake_state import has_workout_skip_today from screen_locker._weekly_check import ( + COUNTED_WORKOUT_TYPES, WEEKLY_WORKOUT_MINIMUM, has_weekly_minimum, is_relaxed_day, @@ -86,6 +88,7 @@ class ScreenLocker( WindowSetupMixin, ShutdownMixin, PhoneVerificationMixin, + RunnerUpVerificationMixin, SickDialogMixin, UIFlowsMixin, UIFlowsRelaxedMixin, @@ -133,6 +136,8 @@ class ScreenLocker( self.container = tk.Frame(self.root, bg="#1a1a1a") self.container.place(relx=0.5, rely=0.5, anchor="center") self._phone_future: Future[tuple[str, str]] | None = None + self._runnerup_future: Future[tuple[str, str]] | None = None + self._runnerup_on_failure: "Callable[[], None] | None" = None if verify_only: self._start_verify_workout_check() elif self._relaxed_day_mode: @@ -217,26 +222,69 @@ class ScreenLocker( return def _try_auto_upgrade_sick_day(self) -> bool: - """Silently upgrade today's sick_day entry if phone shows a workout.""" + """Silently upgrade today's sick_day entry if phone or RunnerUp shows a workout.""" try: status, message = self._verify_phone_workout() except (OSError, RuntimeError) as exc: _logger.info("Auto-upgrade phone check failed: %s", exc) + status, message = "error", str(exc) + if status == "verified": + self.workout_data["type"] = "phone_verified" + self.workout_data["source"] = message + self.workout_data["after_sick_day"] = "true" + self._adjust_shutdown_time_later() + self.save_workout_log() + return True + _logger.info("Auto-upgrade phone skipped (%s), trying RunnerUp...", status) + try: + runnerup_status, runnerup_msg = self._verify_runnerup_workout() + except (OSError, RuntimeError) as exc: + _logger.info("Auto-upgrade RunnerUp check failed: %s", exc) return False - if status != "verified": + if runnerup_status != "verified": _logger.info( - "Auto-upgrade skipped (phone status=%s): %s", - status, - message, + "Auto-upgrade RunnerUp skipped (%s): %s", runnerup_status, runnerup_msg ) return False - self.workout_data["type"] = "phone_verified" - self.workout_data["source"] = message + self.workout_data["type"] = "runnerup_verified" + self.workout_data["source"] = runnerup_msg self.workout_data["after_sick_day"] = "true" self._adjust_shutdown_time_later() self.save_workout_log() return True + def _try_auto_upgrade_early_bird(self) -> bool: + """Override: try phone then RunnerUp to upgrade an early_bird entry.""" + try: + status, message = self._verify_phone_workout() + except (OSError, RuntimeError) as exc: + _logger.info("Early bird upgrade phone check failed: %s", exc) + status, message = "error", str(exc) + if status == "verified": + self.workout_data["type"] = "phone_verified" + self.workout_data["source"] = message + self.workout_data["after_early_bird"] = "true" + self._adjust_shutdown_time_later() + self.save_workout_log() + return True + _logger.info("Early bird phone skipped (%s), trying RunnerUp...", status) + try: + runnerup_status, runnerup_msg = self._verify_runnerup_workout() + except (OSError, RuntimeError) as exc: + _logger.info("Early bird RunnerUp check failed: %s", exc) + return False + if runnerup_status != "verified": + _logger.info( + "Early bird RunnerUp skipped (%s): %s", runnerup_status, runnerup_msg + ) + return False + self.workout_data["type"] = "runnerup_verified" + self.workout_data["source"] = runnerup_msg + self.workout_data["after_early_bird"] = "true" + self._adjust_shutdown_time_later() + self.save_workout_log() + return True + # ------------------------------------------------------------------ # Unlock, logging # ------------------------------------------------------------------ @@ -244,11 +292,11 @@ class ScreenLocker( def _try_adjust_shutdown_for_workout(self) -> bool: """Try to adjust shutdown time later for actual workouts.""" workout_type = self.workout_data.get("type", "") - if workout_type != "phone_verified": + if workout_type not in COUNTED_WORKOUT_TYPES: return False adjusted = self._adjust_shutdown_time_later() if adjusted: - _logger.info("Shutdown time moved 1.5 hours later as workout reward") + _logger.info("Shutdown time moved 2 hours later as workout reward") return adjusted def _clear_debt_on_verified_workout(self) -> int | None: @@ -257,7 +305,7 @@ class ScreenLocker( Returns the new debt count, or ``None`` when this wasn't a phone-verified workout. """ - if self.workout_data.get("type") != "phone_verified": + if self.workout_data.get("type") not in ("phone_verified", "runnerup_verified"): return None history = _sick_tracker.load_history() if history.debt <= 0: @@ -275,7 +323,7 @@ class ScreenLocker( self._label("Great job! 💪", font_size=48, color="#00ff00", pady=30) if shutdown_adjusted: self._text( - "Shutdown time +1.5h later! 🎁", + "Shutdown time +2h later! 🎁", font_size=24, color="#ffaa00", ) @@ -286,7 +334,7 @@ class ScreenLocker( color="#ffaa00" if new_debt > 0 else "#888888", ) self._text("Screen Unlocked!", font_size=36, pady=20) - if self.workout_data.get("type") == "phone_verified": + if self.workout_data.get("type") in ("phone_verified", "runnerup_verified"): self.root.after( 1500, lambda: self._show_commitment_prompt(on_done=self.close),