import os import subprocess import time from typing import Literal, Optional import requests import uvicorn from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from pydantic import BaseModel app = FastAPI() app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) GITHUB_REPO = "ignaciolagosruiz/hevy-sense2-companion" FBA_CACHE = "/tmp/hevy-companion-latest.fba" APK_CACHE = "/tmp/hevy-bridge-latest.apk" STATIC_FBA = "/tmp/hevy-companion-v0.1.fba" STATIC_APK = "/tmp/hevy-bridge.apk" HANDOFF_PATH = "/tmp/handoff-report.md" SCREENSHOT_PATH = "/tmp/hevy-live.png" GAM_ASSET_DIR = "/home/ubuntu/hevy-artifacts/gam" GAM_ASSETS_ZIP = "/home/ubuntu/hevy-artifacts/gam-assets.zip" _last_state = { "phase": "idle", "exercise": "Start workout in Hevy", "setCurrent": 0, "setTotal": 0, "weight": "", "reps": "", "restSeconds": 0, "duration": "", "volume": "", "timestamp": time.time(), "source": "cloud-debug-backend", } class ActionPayload(BaseModel): action: Literal["set_done", "skip_rest", "timer_adjust"] delta: Optional[int] = None def take_screenshot() -> bool: """Cloud/debug path only. Production gym path uses phone-bridge AccessibilityService.""" try: result = subprocess.run(["adb", "exec-out", "screencap", "-p"], capture_output=True, timeout=10) if result.returncode == 0 and len(result.stdout) > 1000: with open(SCREENSHOT_PATH, "wb") as f: f.write(result.stdout) return True except Exception as e: print(f"Screenshot error: {e}") return False @app.get("/api/hevy/live") async def hevy_live(): take_screenshot() _last_state["timestamp"] = time.time() return _last_state @app.get("/api/hevy/update") async def hevy_update( phase: str = "", exercise: str = "", setCurrent: int = -1, setTotal: int = -1, weight: str = "", reps: str = "", restSeconds: int = -1, duration: str = "", volume: str = "", ): if phase: _last_state["phase"] = phase if exercise: _last_state["exercise"] = exercise if setCurrent >= 0: _last_state["setCurrent"] = setCurrent if setTotal >= 0: _last_state["setTotal"] = setTotal if weight: _last_state["weight"] = weight if reps: _last_state["reps"] = reps if restSeconds >= 0: _last_state["restSeconds"] = restSeconds if duration: _last_state["duration"] = duration if volume: _last_state["volume"] = volume _last_state["timestamp"] = time.time() return {"ok": True, "state": _last_state} @app.post("/api/hevy/action") async def hevy_action(payload: ActionPayload): """Remote debug fallback. Real gym actions are handled by phone-bridge local HTTP.""" if payload.action == "set_done": _last_state["phase"] = "rest" _last_state["restSeconds"] = max(1, int(_last_state.get("restSeconds") or 90)) elif payload.action == "skip_rest": _last_state["phase"] = "exercise" _last_state["restSeconds"] = 0 elif payload.action == "timer_adjust": delta = payload.delta or 0 _last_state["restSeconds"] = max(0, int(_last_state.get("restSeconds", 0)) + delta) _last_state["timestamp"] = time.time() return {"ok": True, "action": payload.action, "state": _last_state, "note": "cloud debug fallback only"} def github_headers() -> dict: headers = {"Accept": "application/vnd.github+json"} token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") if token: headers["Authorization"] = f"Bearer {token}" return headers def fetch_latest_asset(suffix: str, cache_path: str): try: headers = github_headers() r = requests.get( f"https://api.github.com/repos/{GITHUB_REPO}/releases", headers=headers, timeout=10, ) r.raise_for_status() for release in r.json(): for asset in release.get("assets", []): if asset["name"].endswith(suffix): asset_headers = dict(headers) asset_headers["Accept"] = "application/octet-stream" asset_r = requests.get(asset["url"], headers=asset_headers, timeout=30) asset_r.raise_for_status() with open(cache_path, "wb") as f: f.write(asset_r.content) print(f"Downloaded {asset['name']} from release {release['tag_name']}: {len(asset_r.content)} bytes") return cache_path except Exception as e: print(f"Failed to fetch latest {suffix}: {e}") return None @app.get("/download/app.fba") async def download_fba(): path = fetch_latest_asset(".fba", FBA_CACHE) or FBA_CACHE if not os.path.exists(path): path = STATIC_FBA if not os.path.exists(path): raise HTTPException(404, "No .fba available") return FileResponse(path, filename="hevy-companion.fba", media_type="application/octet-stream") @app.get("/download/hevy-bridge.apk") async def download_apk(): path = fetch_latest_asset(".apk", APK_CACHE) or APK_CACHE if not os.path.exists(path): path = STATIC_APK if not os.path.exists(path): raise HTTPException(404, "No HevyBridge APK available") return FileResponse(path, filename="hevy-bridge.apk", media_type="application/vnd.android.package-archive") @app.get("/download/handoff") async def download_handoff(): if not os.path.exists(HANDOFF_PATH): raise HTTPException(404, "Handoff not found") return FileResponse(HANDOFF_PATH, filename="handoff-report.md", media_type="text/markdown") @app.get("/download/gam-assets.zip") async def download_gam_assets_zip(): if not os.path.exists(GAM_ASSETS_ZIP): raise HTTPException(404, "GAM assets zip not found") return FileResponse(GAM_ASSETS_ZIP, filename="hevy-companion-gam-assets.zip", media_type="application/zip") @app.get("/download/gam/{filename}") async def download_gam_asset(filename: str): allowed = { "icon-80.png", "icon-160.png", "icon-512.png", "screenshot-active-336.png", "screenshot-rest-336.png", } if filename not in allowed: raise HTTPException(404, "GAM asset not found") path = os.path.join(GAM_ASSET_DIR, filename) if not os.path.exists(path): raise HTTPException(404, "GAM asset missing on disk") return FileResponse(path, filename=filename, media_type="image/png") @app.get("/") async def root(): return { "service": "hevy-companion-backend", "mode": "cloud-debug-and-downloads", "gym_path": "Fitbit companion -> http://127.0.0.1:18090 -> HevyBridge APK -> Hevy", "endpoints": ["/api/hevy/live", "/api/hevy/action", "/download/app.fba", "/download/hevy-bridge.apk", "/download/handoff", "/download/gam-assets.zip", "/download/gam/{filename}"], } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8420)